Added db_mysql_live which allows Anope to pull data

from the four main SQL tables in realtime, which
effectively gives us "live" SQL.
Changed eventfd pipe engine to not use buffered write.
Added TryLock to threading engines.
Made blocking SQL queries in our SQL API thread-safe.

16 files changed, 422 insertions, 21 deletions

diff --git a/data/example.conf b/data/example.conf
index 174a0f8a8..561504cb8 100644
--- a/data/example.conf
+++ b/data/example.conf
@@ -275,6 +275,17 @@ options
 	 * Supported:
 	 *  - db_plain
 	 *  - db_mysql
+	 *  - db_mysql_live
+	 *
+	 * The db_mysql_live module is an extension to db_mysql, and should only be used if
+	 * db_mysql is being used. This module pulls data in real time from SQL as it is
+	 * requested by the core as a result of someone executing commands. This effectively
+	 * allows you to edit your database and have it be immediatly refelected back on Anope.
+	 * It is highly recommended you only use this module if your databases is located
+	 * locally as this module will generate many queries per command.
+	 * db_mysql_live only uses threads for commands and non-blocking queries, so it is safe to
+	 * use on large networks without worrying about response time.
+	 *
 	 */
 	database = "db_plain"
 
diff --git a/data/mysql/tables.sql b/data/mysql/tables.sql
index d0f16b0d5..2cda6384e 100644
--- a/data/mysql/tables.sql
+++ b/data/mysql/tables.sql
@@ -270,7 +270,7 @@ CREATE TABLE IF NOT EXISTS `anope_ns_core` (
   `email` text NOT NULL,
   `greet` text NOT NULL,
   `flags` text NOT NULL,
-  `language` smallint(5) unsigned NOT NULL DEFAULT '0',
+  `language` varchar(5) NOT NULL DEFAULT '',
   `channelcount` smallint(5) unsigned NOT NULL DEFAULT '0',
   `memomax` smallint(5) unsigned NOT NULL DEFAULT '0',
   PRIMARY KEY (`display`)
diff --git a/include/modules.h b/include/modules.h
index 0a92caf74..3d1c2f828 100644
--- a/include/modules.h
+++ b/include/modules.h
@@ -447,6 +447,11 @@ class CoreExport Module : public Extensible
 	 */
 	virtual void OnBadWordDel(ChannelInfo *ci, BadWord *bw) { }
 
+	/** Called in findbot()
+	 * @param nick The nick being looked up
+	 */
+	virtual void OnFindBot(const Anope::string &nick) { }
+
 	/** Called before a bot kicks a user
 	 * @param bi The bot sending the kick
 	 * @param c The channel the user is being kicked on
@@ -816,6 +821,11 @@ class CoreExport Module : public Extensible
 	 */
 	virtual void OnChanInfo(User *u, ChannelInfo *ci, bool ShowHidden) { }
 
+	/** Called on cs_findchan()
+	 * @param chname The name being looked up
+	 */
+	virtual void OnFindChan(const Anope::string &chname) { }
+
 	/** Called when a nick is dropped
 	 * @param nick The nick
 	 */
@@ -907,6 +917,18 @@ class CoreExport Module : public Extensible
 	 */
 	virtual void OnNickInfo(User *u, NickAlias *na, bool ShowHidden) { }
 
+	/** Called in findnick()
+	 * Useful to modify the na returned by findnick()
+	 * @param nick The nick being looked up
+	 */
+	virtual void OnFindNick(const Anope::string &nick) { }
+
+	/** Called in findcore()
+	 * Useful to modify the nc returned by findcore()
+	 * @param nick The nick being looked up
+	 */
+	virtual void OnFindCore(const Anope::string &nick) { }
+
 	/** Called when we get informed about a users SSL fingerprint
 	 *  when we call this, the fingerprint should already be stored in the user struct
 	 * @param u pointer to the user
@@ -1046,17 +1068,17 @@ enum Implementation
 		I_OnNickRegister, I_OnNickSuspended, I_OnNickUnsuspended,
 		I_OnDelNick, I_OnDelCore, I_OnChangeCoreDisplay,
 		I_OnDelNickRequest, I_OnMakeNickRequest, I_OnNickClearAccess, I_OnNickAddAccess, I_OnNickEraseAccess,
-		I_OnNickInfo, I_OnFingerprint,
+		I_OnNickInfo, I_OnFindNick, I_OnFindCore,
 
 		/* ChanServ */
 		I_OnChanForbidden, I_OnChanSuspend, I_OnChanDrop, I_OnPreChanExpire, I_OnChanExpire, I_OnAccessAdd, I_OnAccessChange,
 		I_OnAccessDel, I_OnAccessClear, I_OnLevelChange, I_OnChanRegistered, I_OnChanUnsuspend, I_OnDelChan, I_OnChannelCreate,
 		I_OnChannelDelete, I_OnAkickAdd, I_OnAkickDel,
-		I_OnChanInfo,
+		I_OnChanInfo, I_OnFindChan,
 
 		/* BotServ */
 		I_OnBotJoin, I_OnBotKick, I_OnBotCreate, I_OnBotChange, I_OnBotDelete, I_OnBotAssign, I_OnBotUnAssign,
-		I_OnUserKicked, I_OnBotFantasy, I_OnBotNoFantasyAccess, I_OnBotBan, I_OnBadWordAdd, I_OnBadWordDel,
+		I_OnUserKicked, I_OnBotFantasy, I_OnBotNoFantasyAccess, I_OnBotBan, I_OnBadWordAdd, I_OnBadWordDel, I_OnFindBot,
 
 		/* HostServ */
 		I_OnSetVhost, I_OnDeleteVhost,
@@ -1066,7 +1088,7 @@ enum Implementation
 
 		/* Users */
 		I_OnPreUserConnect, I_OnUserConnect, I_OnUserNickChange, I_OnUserQuit, I_OnUserLogoff, I_OnPreJoinChannel,
-		I_OnJoinChannel, I_OnPrePartChannel, I_OnPartChannel,
+		I_OnJoinChannel, I_OnPrePartChannel, I_OnPartChannel, I_OnFingerprint,
 
 		/* OperServ */
 		I_OnDefconLevel, I_OnAddAkill, I_OnDelAkill, I_OnExceptionAdd, I_OnExceptionDel,
diff --git a/include/regchannel.h b/include/regchannel.h
index d8cf64509..4be8a50aa 100644
--- a/include/regchannel.h
+++ b/include/regchannel.h
@@ -110,6 +110,7 @@ class CoreExport ChannelInfo : public Extensible, public Flags<ChannelInfoFlag,
 	Anope::string last_topic_setter;	/* Setter */
 	time_t last_topic_time;			/* Time */
 
+	// These two should be using extensible
 	Anope::string forbidby;
 	Anope::string forbidreason;
 
diff --git a/include/threadengine.h b/include/threadengine.h
index 8df288cae..822774c6e 100644
--- a/include/threadengine.h
+++ b/include/threadengine.h
@@ -99,6 +99,12 @@ class CoreExport Mutex
 	/** Unlock the mutex, it must be locked first
 	 */
 	void Unlock();
+
+	/** Attempt to lock the mutex, will return true on success and false on fail
+	 * Does not block
+	 * @return true or false
+	 */
+	bool TryLock();
 };
 
 class CoreExport Condition : public Mutex
diff --git a/modules/extra/db_mysql_live.cpp b/modules/extra/db_mysql_live.cpp
new file mode 100644
index 000000000..a7bc1609c
--- /dev/null
+++ b/modules/extra/db_mysql_live.cpp
@@ -0,0 +1,323 @@
+#include "module.h"
+#include "sql.h"
+
+class CommandMutex;
+static std::list<CommandMutex *> commands;
+static CommandMutex *current_command = NULL;
+
+class CommandMutex : public Thread
+{
+ public:
+	Mutex mutex;
+	Command *command;
+	CommandSource source;
+	std::vector<Anope::string> params;
+
+	CommandMutex() : Thread()
+	{
+		commands.push_back(this);
+		current_command = this;
+	}
+
+	~CommandMutex()
+	{
+		std::list<CommandMutex *>::iterator it = std::find(commands.begin(), commands.end(), this);
+		if (it != commands.end())
+			commands.erase(it);
+		if (this == current_command)
+			current_command = NULL;
+	}
+
+	void Run()
+	{
+		User *u = this->source.u;
+		BotInfo *bi = this->source.owner;
+
+		if (!command->permission.empty() && !u->Account()->HasCommand(command->permission))
+		{
+			u->SendMessage(bi, ACCESS_DENIED);
+			Log(LOG_COMMAND, "denied", bi) << "Access denied for user " << u->GetMask() << " with command " << command;
+		}
+		else
+		{
+			CommandReturn ret = command->Execute(source, params);
+
+			if (ret == MOD_CONT)
+			{
+				FOREACH_MOD(I_OnPostCommand, OnPostCommand(source, command, params));
+			}
+		}
+
+		this->mutex.Unlock();
+	}
+};
+
+class MySQLLiveModule : public Module, public Pipe
+{
+	service_reference<SQLProvider> SQL;
+
+	SQLResult RunQuery(const Anope::string &query)
+	{
+		if (!this->SQL)
+			throw SQLException("Unable to locate SQL reference, is m_mysql loaded and configured correctly?");
+
+		return SQL->RunQuery(query);
+	}
+
+	const Anope::string Escape(const Anope::string &query)
+	{
+		return SQL ? SQL->Escape(query) : query;
+	}
+
+ public:
+	MySQLLiveModule(const Anope::string &modname, const Anope::string &creator) : Module(modname, creator), SQL("mysql/main")
+	{
+		Implementation i[] = { I_OnPreCommand, I_OnFindBot, I_OnFindChan, I_OnFindNick, I_OnFindCore };
+		ModuleManager::Attach(i, this, 5);
+	}
+
+	EventReturn OnPreCommand(CommandSource &source, Command *command, const std::vector<Anope::string> &params)
+	{
+		CommandMutex *cm = new CommandMutex();
+		try
+		{
+			cm->mutex.Lock();
+			cm->command = command;
+			cm->source = source;
+			cm->params = params;
+
+			commands.push_back(cm);
+
+			// Give processing to the command thread
+			Log(LOG_DEBUG_2) << "db_mysql_live: Waiting for command thread " << cm->command->name << " from " << source.u->nick;
+			threadEngine.Start(cm);
+			cm->mutex.Lock();
+		}
+		catch (const CoreException &ex)
+		{
+			delete cm;
+			Log() << "db_mysql_live: Unable to thread for command: " << ex.GetReason();
+
+			return EVENT_CONTINUE;
+		}
+
+		return EVENT_STOP;
+	}
+
+	void OnNotify()
+	{
+		for (std::list<CommandMutex *>::iterator it = commands.begin(), it_end = commands.end(); it != it_end; ++it)
+		{
+			CommandMutex *cm  = *it;
+
+			// Thread engine will pick this up later
+			if (cm->GetExitState())
+				continue;
+
+			Log(LOG_DEBUG_2) << "db_mysql_live: Waiting for command thread " << cm->command->name << " from " << cm->source.u->nick;
+			current_command = cm;
+
+			// Unlock to give processing back to the command thread
+			cm->mutex.Unlock();
+			// Relock to regain processing once the command thread hangs for any reason
+			cm->mutex.Lock();
+
+			current_command = NULL;
+		}
+	}
+
+
+	void OnFindBot(const Anope::string &nick)
+	{
+		if (!current_command)
+			return;
+
+		CommandMutex *cm = current_command;
+
+		// Give it back to the core
+		cm->mutex.Unlock();
+		SQLResult res = this->RunQuery("SELECT * FROM `anope_bs_core` WHERE `nick` = '" + this->Escape(nick) + "'");
+		// And take it back...
+		this->Notify();
+		cm->mutex.Lock();
+
+		try
+		{
+			current_command = NULL;
+			BotInfo *bi = findbot(res.Get(0, "nick"));
+			if (!bi)
+				bi = new BotInfo(res.Get(0, "nick"), res.Get(0, "user"), res.Get(0, "host"), res.Get(0, "rname"));
+			else
+			{
+				bi->SetIdent(res.Get(0, "user"));
+				bi->host = res.Get(0, "host");
+				bi->realname = res.Get(0, "rname");
+			}
+
+			if (res.Get(0, "flags").equals_cs("PRIVATE"))
+				bi->SetFlag(BI_PRIVATE);
+			bi->created = convertTo<time_t>(res.Get(0, "created"));
+			bi->chancount = convertTo<uint32>(res.Get(0, "chancount"));
+		}
+		catch (const SQLException &) { }
+		catch (const ConvertException &) { }
+	}
+
+	void OnFindChan(const Anope::string &chname)
+	{
+		if (!current_command)
+			return;
+
+		CommandMutex *cm = current_command;
+
+		cm->mutex.Unlock();
+		SQLResult res = this->RunQuery("SELECT * FROM `anope_cs_info` WHERE `name` = '" + this->Escape(chname) + "'");
+		this->Notify();
+		cm->mutex.Lock();
+
+		try
+		{
+			current_command = NULL;
+			ChannelInfo *ci = cs_findchan(res.Get(0, "name"));
+			if (!ci)
+				ci = new ChannelInfo(res.Get(0, "name"));
+			ci->founder = findcore(res.Get(0, "founder"));
+			ci->successor = findcore(res.Get(0, "successor"));
+			ci->desc = res.Get(0, "descr");
+			ci->time_registered = convertTo<time_t>(res.Get(0, "time_registered"));
+			// XXX flags, we need ChannelInfo::ProcessFlags or similar?
+			ci->forbidby = res.Get(0, "forbidby");
+			ci->forbidreason = res.Get(0, "forbidreason");
+			ci->bantype = convertTo<int>(res.Get(0, "bantype"));
+			ci->memos.memomax = convertTo<unsigned>(res.Get(0, "memomax"));
+
+			Anope::string mlock_on = res.Get(0, "mlock_on"),
+					mlock_off = res.Get(0, "mlock_off"),
+					mlock_params = res.Get(0, "mlock_params"),
+					mlock_params_off = res.Get(0, "mlock_params_off");
+
+			Anope::string mode;
+			std::vector<Anope::string> modes;
+
+			spacesepstream sep(mlock_on);
+			while (sep.GetToken(mode))
+				modes.push_back(mode);
+			ci->Extend("db_mlock_modes_on", new ExtensibleItemRegular<std::vector<Anope::string> >(modes));
+
+			modes.clear();
+			sep = mlock_off;
+			while (sep.GetToken(mode))
+				modes.push_back(mode);
+			ci->Extend("db_mlock_modes_off", new ExtensibleItemRegular<std::vector<Anope::string> >(modes));
+
+			modes.clear();
+			sep = mlock_params;
+			while (sep.GetToken(mode))
+				modes.push_back(mode);
+			ci->Extend("mlock_params", new ExtensibleItemRegular<std::vector<Anope::string> >(modes));
+
+			modes.clear();
+			sep = mlock_params_off;
+			while (sep.GetToken(mode))
+				modes.push_back(mode);
+			ci->Extend("mlock_params_off", new ExtensibleItemRegular<std::vector<Anope::string> >(modes));
+
+			ci->LoadMLock();
+
+			if (res.Get(0, "botnick").equals_cs(ci->bi ? ci->bi->nick : "") == false)
+			{
+				if (ci->bi)
+					ci->bi->UnAssign(NULL, ci);
+				BotInfo *bi = findbot(res.Get(0, "botnick"));
+				if (bi)
+					bi->Assign(NULL, ci);
+			}
+
+			ci->capsmin = convertTo<int16>(res.Get(0, "capsmin"));
+			ci->capspercent = convertTo<int16>(res.Get(0, "capspercent"));
+			ci->floodlines = convertTo<int16>(res.Get(0, "floodlines"));
+			ci->floodsecs = convertTo<int16>(res.Get(0, "floodsecs"));
+			ci->repeattimes = convertTo<int16>(res.Get(0, "repeattimes"));
+
+			if (ci->c)
+				check_modes(ci->c);
+		}
+		catch (const SQLException &) { }
+		catch (const ConvertException &) { }
+	}
+
+	void OnFindNick(const Anope::string &nick)
+	{
+		if (!current_command)
+			return;
+
+		CommandMutex *cm = current_command;
+
+		cm->mutex.Unlock();
+		SQLResult res = this->RunQuery("SELECT * FROM `anope_ns_alias` WHERE `nick` = '" + this->Escape(nick) + "'");
+		this->Notify();
+		cm->mutex.Lock();
+
+		try
+		{
+			// Make OnFindCore trigger and look up the core too
+			NickCore *nc = findcore(res.Get(0, "display"));
+			if (!nc)
+				return;
+			current_command = NULL;
+			NickAlias *na = findnick(res.Get(0, "nick"));
+			if (!na)
+				na = new NickAlias(res.Get(0, "nick"), nc);
+
+			na->last_quit = res.Get(0, "last_quit");
+			na->last_realname = res.Get(0, "last_realname");
+			na->last_usermask = res.Get(0, "last_usermask");
+			na->time_registered = convertTo<time_t>(res.Get(0, "time_registered"));
+			na->last_seen = convertTo<time_t>(res.Get(0, "last_seen"));
+			// XXX flags
+
+			if (na->nc != nc)
+			{
+				std::list<NickAlias *>::iterator it = std::find(na->nc->aliases.begin(), na->nc->aliases.end(), na);
+				if (it != na->nc->aliases.end())
+					na->nc->aliases.erase(it);
+
+				na->nc = nc;
+				na->nc->aliases.push_back(na);
+			}
+		}
+		catch (const SQLException &) { }
+		catch (const ConvertException &) { }
+	}
+
+	void OnFindCore(const Anope::string &nick)
+	{
+		if (!current_command)
+			return;
+
+		CommandMutex *cm = current_command;
+
+		cm->mutex.Unlock();
+		SQLResult res = this->RunQuery("SELECT * FROM `anope_ns_core` WHERE `name` = '" + this->Escape(nick) + "'");
+		this->Notify();
+		cm->mutex.Lock();
+
+		try
+		{
+  			current_command = NULL;
+ 			NickCore *nc = findcore(res.Get(0, "display"));
+			if (!nc)
+				nc = new NickCore(res.Get(0, "display"));
+
+			nc->pass = res.Get(0, "pass");
+			nc->email = res.Get(0, "email");
+			nc->greet = res.Get(0, "greet");
+			// flags
+			nc->language = res.Get(0, "language");
+		}
+		catch (const SQLException &) { }
+		catch (const ConvertException &) { }
+	}
+};
+
+MODULE_INIT(MySQLLiveModule)
diff --git a/modules/extra/m_mysql.cpp b/modules/extra/m_mysql.cpp
index eba2c0382..12b4dbcc5 100644
--- a/modules/extra/m_mysql.cpp
+++ b/modules/extra/m_mysql.cpp
@@ -10,7 +10,7 @@
  * This module spawns a single thread that is used to execute blocking MySQL queries.
  * When a module requests a query to be executed it is added to a list for the thread
  * (which never stops looping and sleeing) to pick up and execute, the result of which
- * is inserted in to another queue to be picked up my the main thread. The main thread
+ * is inserted in to another queue to be picked up by the main thread. The main thread
  * uses Pipe to become notified through the socket engine when there are results waiting
  * to be sent back to the modules requesting the query
  */
@@ -309,15 +309,19 @@ void MySQLService::Run(SQLInterface *i, const Anope::string &query)
 
 SQLResult MySQLService::RunQuery(const Anope::string &query)
 {
+	this->Lock.Lock();
 	if (this->CheckConnection() && !mysql_real_query(this->sql, query.c_str(), query.length()))
 	{
 		MYSQL_RES *res = mysql_use_result(this->sql);
 
+		this->Lock.Unlock();
 		return MySQLResult(query, res);
 	}
 	else
 	{
-		return MySQLResult(query, mysql_error(this->sql));
+		Anope::string error = mysql_error(this->sql);
+		this->Lock.Unlock();
+		return MySQLResult(query, error);
 	}
 }
 
@@ -370,9 +374,7 @@ void DispatcherThread::Run()
 			QueryRequest &r = me->QueryRequests.front();
 			this->Unlock();
 
-			r.service->Lock.Lock();
 			SQLResult sresult = r.service->RunQuery(r.query);
-			r.service->Lock.Unlock();
 
 			this->Lock();
 			if (!me->QueryRequests.empty() && me->QueryRequests.front().query == r.query)
@@ -402,7 +404,7 @@ void MySQLPipe::OnNotify()
 		const QueryResult &qr = *it;
 
 		if (!qr.sqlinterface)
-			throw SQLException("NULL qr.interface in MySQLPipe::OnNotify() ?");
+			throw SQLException("NULL qr.sqlinterface in MySQLPipe::OnNotify() ?");
 
 		if (qr.result.GetError().empty())
 			qr.sqlinterface->OnResult(qr.result);
diff --git a/modules/socketengines/m_socketengine_poll.cpp b/modules/socketengines/m_socketengine_poll.cpp
index 5b44f8674..bca06fe93 100644
--- a/modules/socketengines/m_socketengine_poll.cpp
+++ b/modules/socketengines/m_socketengine_poll.cpp
@@ -162,7 +162,7 @@ class SocketEnginePoll : public SocketEngineBase
 				s->SetFlag(SF_DEAD);
 		}
 
-		for (int i = 0; i < total; ++i)
+		for (int i = 0; i < SocketCount; ++i)
 		{
 			pollfd *ev = &this->events[i];
 			Socket *s = Sockets[ev->fd];
diff --git a/src/botserv.cpp b/src/botserv.cpp
index c96ccb536..23482f76c 100644
--- a/src/botserv.cpp
+++ b/src/botserv.cpp
@@ -359,10 +359,15 @@ void botchanmsgs(User *u, ChannelInfo *ci, const Anope::string &buf)
 
 BotInfo *findbot(const Anope::string &nick)
 {
+	BotInfo *bi;
 	if (isdigit(nick[0]) && ircd->ts6)
-		return BotListByUID.find(nick);
-
-	return BotListByNick.find(nick);
+		bi = BotListByUID.find(nick);
+	else
+		bi = BotListByNick.find(nick);
+	
+	FOREACH_MOD(I_OnFindBot, OnFindBot(nick));
+	
+	return bi;
 }
 
 /*************************************************************************/
diff --git a/src/chanserv.cpp b/src/chanserv.cpp
index b66a36dbe..ef203c064 100644
--- a/src/chanserv.cpp
+++ b/src/chanserv.cpp
@@ -471,10 +471,12 @@ void cs_remove_nick(NickCore *nc)
 
 ChannelInfo *cs_findchan(const Anope::string &chan)
 {
-	registered_channel_map::const_iterator it = RegisteredChannelList.find(chan);
+	FOREACH_MOD(I_OnFindChan, OnFindChan(chan));
 
+	registered_channel_map::const_iterator it = RegisteredChannelList.find(chan);
 	if (it != RegisteredChannelList.end())
 		return it->second;
+
 	return NULL;
 }
 
diff --git a/src/init.cpp b/src/init.cpp
index c12f600fc..fc53d6a2b 100644
--- a/src/init.cpp
+++ b/src/init.cpp
@@ -375,13 +375,13 @@ void Init(int ac, char **av)
 	/* Add Encryption Modules */
 	ModuleManager::LoadModuleList(Config->EncModuleList);
 
-	/* Add Database Modules */
-	ModuleManager::LoadModuleList(Config->DBModuleList);
-
 	/* Load the socket engine */
 	if (ModuleManager::LoadModule(Config->SocketEngine, NULL) || !SocketEngine)
 		throw FatalException("Unable to load socket engine " + Config->SocketEngine);
 
+	/* Add Database Modules */
+	ModuleManager::LoadModuleList(Config->DBModuleList);
+
 	try
 	{
 		DNSEngine = new DNSManager();
diff --git a/src/nickserv.cpp b/src/nickserv.cpp
index 65ffc02d1..f841ced10 100644
--- a/src/nickserv.cpp
+++ b/src/nickserv.cpp
@@ -319,10 +319,12 @@ NickRequest *findrequestnick(const Anope::string &nick)
 
 NickAlias *findnick(const Anope::string &nick)
 {
-	nickalias_map::const_iterator it = NickAliasList.find(nick);
+	FOREACH_MOD(I_OnFindNick, OnFindNick(nick));
 
+	nickalias_map::const_iterator it = NickAliasList.find(nick);
 	if (it != NickAliasList.end())
 		return it->second;
+
 	return NULL;
 }
 
@@ -330,10 +332,12 @@ NickAlias *findnick(const Anope::string &nick)
 
 NickCore *findcore(const Anope::string &nick)
 {
-	nickcore_map::const_iterator it = NickCoreList.find(nick);
+	FOREACH_MOD(I_OnFindCore, OnFindCore(nick));
 
+	nickcore_map::const_iterator it = NickCoreList.find(nick);
 	if (it != NickCoreList.end())
 		return it->second;
+
 	return NULL;
 }
 
diff --git a/src/regchannel.cpp b/src/regchannel.cpp
index 4ab4062aa..b240b8554 100644
--- a/src/regchannel.cpp
+++ b/src/regchannel.cpp
@@ -494,6 +494,8 @@ void ChannelInfo::ClearBadWords()
  */
 void ChannelInfo::LoadMLock()
 {
+	this->ClearMLock();
+
 	std::vector<Anope::string> modenames_on, modenames_off;
 
 	// Force +r
diff --git a/src/socketengines/socketengine_eventfd.cpp b/src/socketengines/socketengine_eventfd.cpp
index 4e4016326..9e9ec677b 100644
--- a/src/socketengines/socketengine_eventfd.cpp
+++ b/src/socketengines/socketengine_eventfd.cpp
@@ -53,7 +53,12 @@ bool Pipe::Read(const Anope::string &)
 
 void Pipe::Notify()
 {
-	this->Write("*");
+	/* Note we send this immediatly. If use use Socket::Write and if this functions is called
+	 * from a thread, only epoll is able to pick up the change to this sockets want flags immediately
+	 * Other engines time out then pick up and write the change then read it back, which
+	 * is too slow for most things.
+	 */
+	this->IO->Send(this, "");
 }
 
 void Pipe::OnNotify()
diff --git a/src/threadengines/threadengine_pthread.cpp b/src/threadengines/threadengine_pthread.cpp
index 65d1533f3..acdccfeb5 100644
--- a/src/threadengines/threadengine_pthread.cpp
+++ b/src/threadengines/threadengine_pthread.cpp
@@ -79,6 +79,15 @@ void Mutex::Unlock()
 	pthread_mutex_unlock(&mutex);
 }
 
+/** Attempt to lock the mutex, will return true on success and false on fail
+ * Does not block
+ * @return true or false
+ */
+bool Mutex::TryLock()
+{
+	return pthread_mutex_trylock(&mutex) == 0;
+}
+
 /** Constructor
  */
 Condition::Condition() : Mutex()
diff --git a/src/threadengines/threadengine_win32.cpp b/src/threadengines/threadengine_win32.cpp
index b270a895d..e6ad725bf 100644
--- a/src/threadengines/threadengine_win32.cpp
+++ b/src/threadengines/threadengine_win32.cpp
@@ -73,6 +73,15 @@ void Mutex::Unlock()
 	LeaveCriticalSection(&mutex);
 }
 
+/** Attempt to lock the mutex, will return true on success and false on fail
+ * Does not block
+ * @return true or false
+ */
+bool Mutex::TryLock()
+{
+	return TryEnterCriticalSection(&mutex);
+}
+
 /** Constructor
  */
 Condition::Condition() : Mutex()