Add multi-Emerge thread support

8 files changed, 160 insertions, 64 deletions

diff --git a/src/defaultsettings.cpp b/src/defaultsettings.cpp
index f18e9b1e0..219cda9e7 100644
--- a/src/defaultsettings.cpp
+++ b/src/defaultsettings.cpp
@@ -187,6 +187,7 @@ void set_default_settings(Settings *settings)
 	settings->setDefault("emergequeue_limit_total", "256");
 	settings->setDefault("emergequeue_limit_diskonly", "5");
 	settings->setDefault("emergequeue_limit_generate", "1");
+	settings->setDefault("num_emerge_threads", "");
 	
 	// physics stuff
 	settings->setDefault("movement_acceleration_default", "3");
diff --git a/src/emerge.cpp b/src/emerge.cpp
index b785c8688..ee6650f9c 100644
--- a/src/emerge.cpp
+++ b/src/emerge.cpp
@@ -47,49 +47,56 @@ EmergeManager::EmergeManager(IGameDef *gamedef, BiomeDefManager *bdef) {
 
 	this->biomedef = bdef ? bdef : new BiomeDefManager(gamedef);
 	this->params   = NULL;
-	this->mapgen   = NULL;
 	
 	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();
-	emergethread = new EmergeThread((Server *)gamedef);
+	int nthreads = g_settings->get("num_emerge_threads").empty() ?
+					porting::getNumberOfProcessors() :
+					g_settings->getU16("num_emerge_threads");
+	if (nthreads < 1)
+		nthreads = 1;
+	
+	for (int i = 0; i != nthreads; i++)
+		emergethread.push_back(new EmergeThread((Server *)gamedef, i));
+		
+	infostream << "EmergeManager: using " << nthreads << " threads" << std::endl;
 }
 
 
 EmergeManager::~EmergeManager() {
-	emergethread->setRun(false);
-	emergethread->qevent.signal();
-	emergethread->stop();
+	for (int i = 0; i != emergethread.size(); i++) {
+		emergethread[i]->setRun(false);
+		emergethread[i]->qevent.signal();
+		emergethread[i]->stop();
+		delete emergethread[i];
+		delete mapgen[i];
+	}
 	
-	delete emergethread;
 	delete biomedef;
-	delete mapgen;
 	delete params;
 }
 
 
 void EmergeManager::initMapgens(MapgenParams *mgparams) {
-	if (mapgen)
+	Mapgen *mg;
+	
+	if (mapgen.size())
 		return;
 	
 	this->params = mgparams;
-	this->mapgen = getMapgen(); //only one mapgen for now!
-}
-
-
-Mapgen *EmergeManager::getMapgen() {
-	if (!mapgen) {
-		mapgen = createMapgen(params->mg_name, 0, params, this);
-		if (!mapgen) {
+	for (int i = 0; i != emergethread.size(); i++) {
+		mg = createMapgen(params->mg_name, 0, params);
+		if (!mg) {
 			infostream << "EmergeManager: falling back to mapgen v6" << std::endl;
 			delete params;
 			params = createMapgenParams("v6");
-			mapgen = createMapgen("v6", 0, params, this);
+			mg = createMapgen("v6", 0, params);
 		}
+		mapgen.push_back(mg);
 	}
-	return mapgen;
 }
 
 
@@ -98,6 +105,7 @@ bool EmergeManager::enqueueBlockEmerge(u16 peer_id, v3s16 p, bool allow_generate
 	BlockEmergeData *bedata;
 	u16 count;
 	u8 flags = 0;
+	int idx = 0;
 	
 	if (allow_generate)
 		flags |= BLOCK_EMERGE_ALLOWGEN;
@@ -128,45 +136,58 @@ bool EmergeManager::enqueueBlockEmerge(u16 peer_id, v3s16 p, bool allow_generate
 		
 		peer_queue_count[peer_id] = count + 1;
 		
-		emergethread->blockqueue.push(p);
+		int lowestitems = emergethread[0]->blockqueue.size();
+		for (int i = 1; i != emergethread.size(); i++) {
+			int nitems = emergethread[i]->blockqueue.size();
+			if (nitems < lowestitems) {
+				idx = i;
+				lowestitems = nitems;
+			}
+		}
+		
+		emergethread[idx]->blockqueue.push(p);
 	}
-	emergethread->qevent.signal();
+	emergethread[idx]->qevent.signal();
 	
 	return true;
 }
 
 
-bool EmergeManager::popBlockEmerge(v3s16 *pos, u8 *flags) {
+bool EmergeThread::popBlockEmerge(v3s16 *pos, u8 *flags) {
 	std::map<v3s16, BlockEmergeData *>::iterator iter;
-	JMutexAutoLock queuelock(queuemutex);
+	JMutexAutoLock queuelock(emerge->queuemutex);
 
-	if (emergethread->blockqueue.empty())
+	if (blockqueue.empty())
 		return false;
-	v3s16 p = emergethread->blockqueue.front();
-	emergethread->blockqueue.pop();
+	v3s16 p = blockqueue.front();
+	blockqueue.pop();
 	
 	*pos = p;
 	
-	iter = blocks_enqueued.find(p);
-	if (iter == blocks_enqueued.end()) 
+	iter = emerge->blocks_enqueued.find(p);
+	if (iter == emerge->blocks_enqueued.end()) 
 		return false; //uh oh, queue and map out of sync!!
 
 	BlockEmergeData *bedata = iter->second;
 	*flags = bedata->flags;
 	
-	peer_queue_count[bedata->peer_requested]--;
+	emerge->peer_queue_count[bedata->peer_requested]--;
 
 	delete bedata;
-	blocks_enqueued.erase(iter);
+	emerge->blocks_enqueued.erase(iter);
 	
 	return true;
 }
 
 
 int EmergeManager::getGroundLevelAtPoint(v2s16 p) {
-	if (!mapgen)
+	if (!mapgen[0]) {
+		errorstream << "EmergeManager: getGroundLevelAtPoint() called"
+		" before mapgen initialized" << std::endl;
 		return 0;
-	return mapgen->getGroundLevelAtPoint(p);
+	}
+	
+	return mapgen[0]->getGroundLevelAtPoint(p);
 }
 
 
@@ -193,8 +214,9 @@ u32 EmergeManager::getBlockSeed(v3s16 p) {
 
 
 Mapgen *EmergeManager::createMapgen(std::string mgname, int mgid,
-									MapgenParams *mgparams, EmergeManager *emerge) {
-	std::map<std::string, MapgenFactory *>::const_iterator iter = mglist.find(mgname);
+									 MapgenParams *mgparams) {
+	std::map<std::string, MapgenFactory *>::const_iterator iter;
+	iter = mglist.find(mgname);
 	if (iter == mglist.end()) {
 		errorstream << "EmergeManager; mapgen " << mgname <<
 		 " not registered" << std::endl;
@@ -202,12 +224,13 @@ Mapgen *EmergeManager::createMapgen(std::string mgname, int mgid,
 	}
 	
 	MapgenFactory *mgfactory = iter->second;
-	return mgfactory->createMapgen(mgid, mgparams, emerge);
+	return mgfactory->createMapgen(mgid, mgparams, this);
 }
 
 
 MapgenParams *EmergeManager::createMapgenParams(std::string mgname) {
-	std::map<std::string, MapgenFactory *>::const_iterator iter = mglist.find(mgname);
+	std::map<std::string, MapgenFactory *>::const_iterator iter;
+	iter = mglist.find(mgname);
 	if (iter == mglist.end()) {
 		errorstream << "EmergeManager: mapgen " << mgname <<
 		 " not registered" << std::endl;
@@ -227,7 +250,7 @@ MapgenParams *EmergeManager::getParamsFromSettings(Settings *settings) {
 	mgparams->seed        = settings->getU64(settings == g_settings ? "fixed_map_seed" : "seed");
 	mgparams->water_level = settings->getS16("water_level");
 	mgparams->chunksize   = settings->getS16("chunksize");
-	mgparams->flags       = settings->getS32("mg_flags");
+	mgparams->flags       = settings->getFlagStr("mg_flags", flagdesc_mapgen);
 
 	if (!mgparams->readParams(settings)) {
 		delete mgparams;
@@ -354,11 +377,11 @@ void *EmergeThread::Thread() {
 	
 	map    = (ServerMap *)&(m_server->m_env->getMap());
 	emerge = m_server->m_emerge;
-	mapgen = emerge->getMapgen();
+	mapgen = emerge->mapgen[id]; //emerge->getMapgen();
 	
 	while (getRun())
 	try {
-		while (!emerge->popBlockEmerge(&p, &flags)) {
+		while (!popBlockEmerge(&p, &flags)) {
 			qevent.wait();
 			if (!getRun())
 				goto exit_emerge_loop;
diff --git a/src/emerge.h b/src/emerge.h
index b4461ae61..7e0cc4850 100644
--- a/src/emerge.h
+++ b/src/emerge.h
@@ -46,8 +46,8 @@ class EmergeManager {
 public:
 	std::map<std::string, MapgenFactory *> mglist;
 	
-	Mapgen *mapgen;
-	EmergeThread *emergethread;
+	std::vector<Mapgen *> mapgen;
+	std::vector<EmergeThread *> emergethread;
 	
 	//settings
 	MapgenParams *params;
@@ -68,11 +68,9 @@ public:
 
 	void initMapgens(MapgenParams *mgparams);
 	Mapgen *createMapgen(std::string mgname, int mgid,
-						MapgenParams *mgparams, EmergeManager *emerge);
+						MapgenParams *mgparams);
 	MapgenParams *createMapgenParams(std::string mgname);
-	Mapgen *getMapgen();
 	bool enqueueBlockEmerge(u16 peer_id, v3s16 p, bool allow_generate);
-	bool popBlockEmerge(v3s16 *pos, u8 *flags);
 	
 	bool registerMapgen(std::string name, MapgenFactory *mgfactory);
 	MapgenParams *getParamsFromSettings(Settings *settings);
@@ -92,17 +90,19 @@ class EmergeThread : public SimpleThread
 	EmergeManager *emerge;
 	Mapgen *mapgen;
 	bool enable_mapgen_debug_info;
+	int id;
 	
 public:
 	Event qevent;
 	std::queue<v3s16> blockqueue;
 	
-	EmergeThread(Server *server):
+	EmergeThread(Server *server, int ethreadid):
 		SimpleThread(),
 		m_server(server),
 		map(NULL),
 		emerge(NULL),
-		mapgen(NULL)
+		mapgen(NULL),
+		id(ethreadid)
 	{
 		enable_mapgen_debug_info = g_settings->getBool("enable_mapgen_debug_info");
 	}
@@ -118,6 +118,7 @@ public:
 		}
 	}
 
+	bool popBlockEmerge(v3s16 *pos, u8 *flags);
 	bool getBlockOrStartGen(v3s16 p, MapBlock **b, 
 							BlockMakeData *data, bool allow_generate);
 };
diff --git a/src/map.cpp b/src/map.cpp
index 7eb45463f..a8928d864 100644
--- a/src/map.cpp
+++ b/src/map.cpp
@@ -2009,7 +2009,7 @@ ServerMap::ServerMap(std::string savedir, IGameDef *gamedef, EmergeManager *emer
 	m_mgparams = m_emerge->getParamsFromSettings(g_settings);
 	if (!m_mgparams)
 		m_mgparams = new MapgenV6Params();
-		
+
 	m_seed = m_mgparams->seed;
 
 	if (g_settings->get("fixed_map_seed").empty())
@@ -2246,6 +2246,21 @@ void ServerMap::initBlockMake(BlockMakeData *data, v3s16 blockpos)
 		//TimeTaker timer("initBlockMake() initialEmerge");
 		data->vmanip->initialEmerge(bigarea_blocks_min, bigarea_blocks_max);
 	}
+	
+	// Ensure none of the blocks to be generated were marked as containing CONTENT_IGNORE
+	for (s16 z = blockpos_min.Z; z <= blockpos_max.Z; z++) {
+		for (s16 y = blockpos_min.Y; y <= blockpos_max.Y; y++) {
+			for (s16 x = blockpos_min.X; x <= blockpos_max.X; x++) {
+				core::map<v3s16, u8>::Node *n;
+				n = data->vmanip->m_loaded_blocks.find(v3s16(x, y, z));
+				if (n == NULL)
+					continue;
+				u8 flags = n->getValue();
+				flags &= ~VMANIP_BLOCK_CONTAINS_CIGNORE;
+				n->setValue(flags);
+			}
+		}
+	}
 
 	// Data is ready now.
 }
@@ -3672,8 +3687,10 @@ void MapVoxelManipulator::emerge(VoxelArea a, s32 caller_id)
 	for(s32 y=p_min.Y; y<=p_max.Y; y++)
 	for(s32 x=p_min.X; x<=p_max.X; x++)
 	{
+		u8 flags = 0;
+		MapBlock *block;
 		v3s16 p(x,y,z);
-		core::map<v3s16, bool>::Node *n;
+		core::map<v3s16, u8>::Node *n;
 		n = m_loaded_blocks.find(p);
 		if(n != NULL)
 			continue;
@@ -3689,7 +3706,7 @@ void MapVoxelManipulator::emerge(VoxelArea a, s32 caller_id)
 			a.print(infostream);
 			infostream<<std::endl;*/
 
-			MapBlock *block = m_map->getBlockNoCreate(p);
+			block = m_map->getBlockNoCreate(p);
 			if(block->isDummy())
 				block_data_inexistent = true;
 			else
@@ -3702,6 +3719,8 @@ void MapVoxelManipulator::emerge(VoxelArea a, s32 caller_id)
 
 		if(block_data_inexistent)
 		{
+			flags |= VMANIP_BLOCK_DATA_INEXIST;
+			
 			VoxelArea a(p*MAP_BLOCKSIZE, (p+1)*MAP_BLOCKSIZE-v3s16(1,1,1));
 			// Fill with VOXELFLAG_INEXISTENT
 			for(s32 z=a.MinEdge.Z; z<=a.MaxEdge.Z; z++)
@@ -3711,8 +3730,13 @@ void MapVoxelManipulator::emerge(VoxelArea a, s32 caller_id)
 				memset(&m_flags[i], VOXELFLAG_INEXISTENT, MAP_BLOCKSIZE);
 			}
 		}
+		else if (block->getNode(0, 0, 0).getContent() == CONTENT_IGNORE)
+		{
+			// Mark that block was loaded as blank
+			flags |= VMANIP_BLOCK_CONTAINS_CIGNORE;
+		}
 
-		m_loaded_blocks.insert(p, !block_data_inexistent);
+		m_loaded_blocks.insert(p, flags);
 	}
 
 	//infostream<<"emerge done"<<std::endl;
@@ -3832,8 +3856,10 @@ void ManualMapVoxelManipulator::initialEmerge(
 	for(s32 y=p_min.Y; y<=p_max.Y; y++)
 	for(s32 x=p_min.X; x<=p_max.X; x++)
 	{
+		u8 flags = 0;
+		MapBlock *block;
 		v3s16 p(x,y,z);
-		core::map<v3s16, bool>::Node *n;
+		core::map<v3s16, u8>::Node *n;
 		n = m_loaded_blocks.find(p);
 		if(n != NULL)
 			continue;
@@ -3843,7 +3869,7 @@ void ManualMapVoxelManipulator::initialEmerge(
 		{
 			TimeTaker timer1("emerge load", &emerge_load_time);
 
-			MapBlock *block = m_map->getBlockNoCreate(p);
+			block = m_map->getBlockNoCreate(p);
 			if(block->isDummy())
 				block_data_inexistent = true;
 			else
@@ -3856,6 +3882,8 @@ void ManualMapVoxelManipulator::initialEmerge(
 
 		if(block_data_inexistent)
 		{
+			flags |= VMANIP_BLOCK_DATA_INEXIST;
+			
 			/*
 				Mark area inexistent
 			*/
@@ -3868,8 +3896,13 @@ void ManualMapVoxelManipulator::initialEmerge(
 				memset(&m_flags[i], VOXELFLAG_INEXISTENT, MAP_BLOCKSIZE);
 			}
 		}
+		else if (block->getNode(0, 0, 0).getContent() == CONTENT_IGNORE)
+		{
+			// Mark that block was loaded as blank
+			flags |= VMANIP_BLOCK_CONTAINS_CIGNORE;
+		}
 
-		m_loaded_blocks.insert(p, !block_data_inexistent);
+		m_loaded_blocks.insert(p, flags);
 	}
 }
 
@@ -3882,12 +3915,14 @@ void ManualMapVoxelManipulator::blitBackAll(
 	/*
 		Copy data of all blocks
 	*/
-	for(core::map<v3s16, bool>::Iterator
+	for(core::map<v3s16, u8>::Iterator
 			i = m_loaded_blocks.getIterator();
 			i.atEnd() == false; i++)
 	{
 		v3s16 p = i.getNode()->getKey();
-		bool existed = i.getNode()->getValue();
+		u8 flags = i.getNode()->getValue();
+		
+		bool existed = !(flags & VMANIP_BLOCK_DATA_INEXIST);
 		if(existed == false)
 		{
 			// The Great Bug was found using this
@@ -3896,6 +3931,7 @@ void ManualMapVoxelManipulator::blitBackAll(
 					<<std::endl;*/
 			continue;
 		}
+		
 		MapBlock *block = m_map->getBlockNoCreateNoEx(p);
 		if(block == NULL)
 		{
@@ -3906,10 +3942,13 @@ void ManualMapVoxelManipulator::blitBackAll(
 			continue;
 		}
 
-		block->copyFrom(*this);
-
-		if(modified_blocks)
-			modified_blocks->insert(p, block);
+		bool no_content_ignore = !(flags & VMANIP_BLOCK_CONTAINS_CIGNORE);
+		if (no_content_ignore)
+		{
+			block->copyFrom(*this);
+			if(modified_blocks)
+				modified_blocks->insert(p, block);
+		}
 	}
 }
 
diff --git a/src/map.h b/src/map.h
index 0b2311f39..1062f8301 100644
--- a/src/map.h
+++ b/src/map.h
@@ -517,6 +517,9 @@ private:
 	sqlite3_stmt *m_database_list;
 };
 
+#define VMANIP_BLOCK_DATA_INEXIST     1
+#define VMANIP_BLOCK_CONTAINS_CIGNORE 2
+
 class MapVoxelManipulator : public VoxelManipulator
 {
 public:
@@ -532,14 +535,14 @@ public:
 	virtual void emerge(VoxelArea a, s32 caller_id=-1);
 
 	void blitBack(core::map<v3s16, MapBlock*> & modified_blocks);
-
-protected:
-	Map *m_map;
+	
 	/*
 		key = blockpos
-		value = block existed when loaded
+		value = flags describing the block
 	*/
-	core::map<v3s16, bool> m_loaded_blocks;
+	core::map<v3s16, u8> m_loaded_blocks;
+protected:
+	Map *m_map;
 };
 
 class ManualMapVoxelManipulator : public MapVoxelManipulator
diff --git a/src/porting.cpp b/src/porting.cpp
index 7ad557833..58d71e4aa 100644
--- a/src/porting.cpp
+++ b/src/porting.cpp
@@ -132,6 +132,29 @@ 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
+}
+
+/*
 	Path mangler
 */
 
diff --git a/src/porting.h b/src/porting.h
index 13b715557..53aad6171 100644
--- a/src/porting.h
+++ b/src/porting.h
@@ -104,6 +104,11 @@ std::string getDataPath(const char *subpath);
 void initializePaths();
 
 /*
+	Get number of online processors in the system.
+*/
+int getNumberOfProcessors();
+
+/*
 	Resolution is 10-20ms.
 	Remember to check for overflows.
 	Overflow can occur at any value higher than 10000000.
diff --git a/src/server.cpp b/src/server.cpp
index f2897d46d..5021718a3 100644
--- a/src/server.cpp
+++ b/src/server.cpp
@@ -1649,7 +1649,8 @@ void Server::AsyncRunStep()
 		{
 			counter = 0.0;
 
-			m_emerge->emergethread->trigger();
+			for (int i = 0; i != m_emerge->emergethread.size(); i++)
+				m_emerge->emergethread[i]->trigger();
 
 			// Update m_enable_rollback_recording here too
 			m_enable_rollback_recording =