2 files changed, 619 insertions, 0 deletions

diff --git a/source/cSocketThreads.cpp b/source/cSocketThreads.cpp
new file mode 100644
index 000000000..f5d007b66
--- /dev/null
+++ b/source/cSocketThreads.cpp
@@ -0,0 +1,472 @@
+

+// cSocketThreads.cpp

+

+// Implements the cSocketThreads class representing the heart of MCS's client networking.

+// This object takes care of network communication, groups sockets into threads and uses as little threads as possible for full read / write support

+// For more detail, see http://forum.mc-server.org/showthread.php?tid=327

+

+#include "Globals.h"

+#include "cSocketThreads.h"

+#include "cClientHandle.h"

+#include "packets/cPacket_RelativeEntityMoveLook.h"

+

+

+

+

+

+///////////////////////////////////////////////////////////////////////////////////////////////////////////////////////

+// cSocketThreads:

+

+cSocketThreads::cSocketThreads(void)

+{

+}

+

+

+

+

+

+cSocketThreads::~cSocketThreads()

+{

+	for (cSocketThreadList::iterator itr = m_Threads.begin(); itr != m_Threads.end(); ++itr)

+	{

+		delete *itr;

+	}  // for itr - m_Threads[]

+	m_Threads.clear();

+}

+

+

+

+

+

+

+void cSocketThreads::AddClient(cSocket * a_Socket, cCallback * a_Client)

+{

+	// Add a (socket, client) pair for processing, data from a_Socket is to be sent to a_Client

+	

+	// Try to add to existing threads:

+	cCSLock Lock(m_CS);

+	for (cSocketThreadList::iterator itr = m_Threads.begin(); itr != m_Threads.end(); ++itr)

+	{

+		if ((*itr)->HasEmptySlot())

+		{

+			(*itr)->AddClient(a_Socket, a_Client);

+			return;

+		}

+	}

+	

+	// No thread has free space, create a new one:

+	cSocketThread * Thread = new cSocketThread(this);

+	Thread->Start();

+	Thread->AddClient(a_Socket, a_Client);

+	m_Threads.push_back(Thread);

+}

+

+

+

+

+

+void cSocketThreads::RemoveClient(cSocket * a_Socket)

+{

+	// Remove the socket (and associated client) from processing

+

+	cCSLock Lock(m_CS);

+	for (cSocketThreadList::iterator itr = m_Threads.begin(); itr != m_Threads.end(); ++itr)

+	{

+		if ((*itr)->RemoveSocket(a_Socket))

+		{

+			return;

+		}

+	}

+}

+

+

+

+

+

+void cSocketThreads::RemoveClient(cCallback * a_Client)

+{

+	// Remove the associated socket and the client from processing

+

+	cCSLock Lock(m_CS);

+	for (cSocketThreadList::iterator itr = m_Threads.begin(); itr != m_Threads.end(); ++itr)

+	{

+		if ((*itr)->RemoveClient(a_Client))

+		{

+			return;

+		}

+	}

+}

+

+

+

+

+

+void cSocketThreads::NotifyWrite(cCallback * a_Client)

+{

+	// Notifies the thread responsible for a_Client that the client has something to write

+

+	cCSLock Lock(m_CS);

+	for (cSocketThreadList::iterator itr = m_Threads.begin(); itr != m_Threads.end(); ++itr)

+	{

+		if ((*itr)->NotifyWrite(a_Client))

+		{

+			return;

+		}

+	}

+}

+

+

+

+

+

+////////////////////////////////////////////////////////////////////////////////

+// cSocketThreads::cSocketThread:

+

+cSocketThreads::cSocketThread::cSocketThread(cSocketThreads * a_Parent) :

+	cIsThread("cSocketThread"),

+	m_Parent(a_Parent),

+	m_NumSlots(0)

+{

+	// Nothing needed yet

+}

+

+

+

+

+

+void cSocketThreads::cSocketThread::AddClient(cSocket * a_Socket, cCallback * a_Client)

+{

+	assert(m_NumSlots < MAX_SLOTS);  // Use HasEmptySlot() to check before adding

+	

+	m_Slots[m_NumSlots].m_Client = a_Client;

+	m_Slots[m_NumSlots].m_Socket = a_Socket;

+	m_Slots[m_NumSlots].m_Outgoing.clear();

+	m_NumSlots++;

+	

+	// Notify the thread of the change:

+	assert(m_ControlSocket2.IsValid());

+	m_ControlSocket2.Send("a", 1);

+}

+

+

+

+

+

+bool cSocketThreads::cSocketThread::RemoveClient(cCallback * a_Client)

+{

+	// Returns true if removed, false if not found

+	

+	if (m_NumSlots == 0)

+	{

+		return false;

+	}

+	

+	for (int i = m_NumSlots - 1; i > 0 ; --i)

+	{

+		if (m_Slots[i].m_Client != a_Client)

+		{

+			continue;

+		}

+		

+		// Found, remove it:

+		m_Slots[i] = m_Slots[m_NumSlots - 1];

+		m_NumSlots--;

+		

+		// Notify the thread of the change:

+		assert(m_ControlSocket2.IsValid());

+		m_ControlSocket2.Send("r", 1);

+		return true;

+	}  // for i - m_Slots[]

+	

+	// Not found

+	return false;

+}

+

+

+

+

+

+bool cSocketThreads::cSocketThread::RemoveSocket(cSocket * a_Socket)

+{

+	// Returns true if removed, false if not found

+

+	if (m_NumSlots == 0)

+	{

+		return false;

+	}

+	

+	for (int i = m_NumSlots - 1; i > 0 ; --i)

+	{

+		if (m_Slots[i].m_Socket != a_Socket)

+		{

+			continue;

+		}

+		

+		// Found, remove it:

+		m_Slots[i] = m_Slots[m_NumSlots - 1];

+		m_NumSlots--;

+		

+		// Notify the thread of the change:

+		assert(m_ControlSocket2.IsValid());

+		m_ControlSocket2.Send("r", 1);

+		return true;

+	}  // for i - m_Slots[]

+	

+	// Not found

+	return false;

+}

+

+

+

+

+

+bool cSocketThreads::cSocketThread::NotifyWrite(cCallback * a_Client)

+{

+	if (HasClient(a_Client))

+	{

+		// Notify the thread that there's another packet in the queue:

+		assert(m_ControlSocket2.IsValid());

+		m_ControlSocket2.Send("q", 1);

+		return true;

+	}

+	return false;

+}

+

+

+

+

+

+bool cSocketThreads::cSocketThread::HasClient(cCallback * a_Client) const

+{

+	for (int i = m_NumSlots - 1; i >= 0; --i)

+	{

+		if (m_Slots[i].m_Client == a_Client)

+		{

+			return true;

+		}

+	}  // for i - m_Slots[]

+	return false;

+}

+

+

+

+

+

+bool cSocketThreads::cSocketThread::HasSocket(cSocket * a_Socket) const

+{

+	for (int i = m_NumSlots - 1; i >= 0; --i)

+	{

+		if (m_Slots[i].m_Socket == a_Socket)

+		{

+			return true;

+		}

+	}  // for i - m_Slots[]

+	return false;

+}

+

+

+

+

+

+bool cSocketThreads::cSocketThread::Start(void)

+{

+	// Create the control socket listener

+	m_ControlSocket1 = cSocket::CreateSocket();

+	if (!m_ControlSocket1.IsValid())

+	{

+		LOGERROR("Cannot create a Control socket for a cSocketThread (\"%s\"); continuing, but server may be unreachable from now on.", cSocket::GetLastErrorString().c_str());

+		return false;

+	}

+	cSocket::SockAddr_In Addr;

+	Addr.Family = cSocket::ADDRESS_FAMILY_INTERNET;

+	Addr.Address = cSocket::INTERNET_ADDRESS_LOCALHOST;

+	Addr.Port = 0;  // Any free port is okay

+	if (m_ControlSocket1.Bind(Addr) != 0)

+	{

+		LOGERROR("Cannot bind a Control socket for a cSocketThread (\"%s\"); continuing, but server may be unreachable from now on.", cSocket::GetLastErrorString().c_str());

+		m_ControlSocket1.CloseSocket();

+		return false;

+	}

+	if (m_ControlSocket1.GetPort() == 0)

+	{

+		LOGERROR("Cannot determine Control socket port (\"%s\"); conitnuing, but the server may be unreachable from now on.", cSocket::GetLastErrorString().c_str());

+		m_ControlSocket1.CloseSocket();

+		return false;

+	}

+

+	// Start the thread

+	if (!super::Start())

+	{

+		m_ControlSocket1.CloseSocket();

+		return false;

+	}

+	

+	// Finish connecting the control socket by accepting connection from the thread's socket

+	cSocket tmp = m_ControlSocket1.Accept();

+	if (!tmp.IsValid())

+	{

+		LOGERROR("Cannot link Control sockets for a cSocketThread (\"%s\"); continuing, but server may be unreachable from now on.", cSocket::GetLastErrorString().c_str());

+		m_ControlSocket1.CloseSocket();

+		return false;

+	}

+	m_ControlSocket1.CloseSocket();

+	m_ControlSocket1 = tmp;

+	

+	return true;

+}

+

+

+

+

+

+void cSocketThreads::cSocketThread::Execute(void)

+{

+	// Connect the "client" part of the Control socket:

+	cSocket::SockAddr_In Addr;

+	Addr.Family = cSocket::ADDRESS_FAMILY_INTERNET;

+	Addr.Address = cSocket::INTERNET_ADDRESS_LOCALHOST;

+	Addr.Port = m_ControlSocket1.GetPort();

+	assert(Addr.Port != 0);  // We checked in the Start() method, but let's be sure

+	if (m_ControlSocket2.Connect(Addr) != 0)

+	{

+		LOGERROR("Cannot connect Control sockets for a cSocketThread (\"%s\"); continuing, but the server may be unreachable from now on.", cSocket::GetLastErrorString().c_str());

+		m_ControlSocket1.CloseSocket();

+		return;

+	}

+	

+	// The main thread loop:

+	while (!mShouldTerminate)

+	{

+		// Put all sockets into the Read set:

+		fd_set fdRead;

+		cSocket::xSocket Highest = 0;

+		

+		PrepareSet(&fdRead, Highest);

+		

+		// Wait for the sockets:

+		if (select(Highest + 1, &fdRead, NULL, NULL, NULL) == -1)

+		{

+			LOGWARNING("select(R) call failed in cSocketThread: \"%s\"", cSocket::GetLastErrorString().c_str());

+			break;

+		}

+		

+		ReadFromSockets(&fdRead);

+

+		// Test sockets for writing:

+		fd_set fdWrite;

+		Highest = 0;

+		PrepareSet(&fdWrite, Highest);		

+		timeval Timeout;

+		Timeout.tv_sec = 0;

+		Timeout.tv_usec = 0;

+		if (select(Highest + 1, NULL, &fdWrite, NULL, &Timeout) == -1)

+		{

+			LOGWARNING("select(W) call failed in cSocketThread: \"%s\"", cSocket::GetLastErrorString().c_str());

+			break;

+		}

+		

+		WriteToSockets(&fdWrite);

+	}  // while (!mShouldTerminate)

+	

+	LOG("cSocketThread %p is terminating", this);

+}

+

+

+

+

+

+void cSocketThreads::cSocketThread::PrepareSet(fd_set * a_Set, cSocket::xSocket & a_Highest)

+{

+	FD_ZERO(a_Set);

+	FD_SET(m_ControlSocket1.GetSocket(), a_Set);

+

+	cCSLock Lock(m_Parent->m_CS);

+	for (int i = m_NumSlots - 1; i > 0; --i)

+	{

+		if (!m_Slots[i].m_Socket->IsValid())

+		{

+			continue;

+		}

+		cSocket::xSocket s = m_Slots[i].m_Socket->GetSocket();

+		FD_SET(s, a_Set);

+		if (s > a_Highest)

+		{

+			a_Highest = s;

+		}

+	}  // for i - m_Slots[]

+}

+

+

+

+

+

+void cSocketThreads::cSocketThread::ReadFromSockets(fd_set * a_Read)

+{

+	// Read on available sockets:

+	cCSLock Lock(m_Parent->m_CS);

+	for (int i = m_NumSlots - 1; i > 0; --i)

+	{

+		if (!FD_ISSET(m_Slots[i].m_Socket->GetSocket(), a_Read))

+		{

+			continue;

+		}

+		char Buffer[1024];

+		int Received = m_Slots[i].m_Socket->Receive(Buffer, ARRAYCOUNT(Buffer), 0);

+		if (Received == 0)

+		{

+			// The socket has been closed by the remote party, close our socket and let it be removed after we process all reading

+			m_Slots[i].m_Socket->CloseSocket();

+			m_Slots[i].m_Client->SocketClosed();

+		}

+		else if (Received > 0)

+		{

+			m_Slots[i].m_Client->DataReceived(Buffer, Received);

+		}

+		else

+		{

+			// The socket has encountered an error, close it and let it be removed after we process all reading

+			m_Slots[i].m_Socket->CloseSocket();

+			m_Slots[i].m_Client->SocketClosed();

+		}

+	}  // for i - m_Slots[]

+}

+

+

+

+

+

+void cSocketThreads::cSocketThread::WriteToSockets(fd_set * a_Write)

+{

+	cCSLock Lock(m_Parent->m_CS);

+	for (int i = m_NumSlots - 1; i > 0; --i)

+	{

+		if (!FD_ISSET(m_Slots[i].m_Socket->GetSocket(), a_Write))

+		{

+			continue;

+		}

+		if (m_Slots[i].m_Outgoing.empty())

+		{

+			// Request another chunk of outgoing data:

+			m_Slots[i].m_Client->GetOutgoingData(m_Slots[i].m_Outgoing);

+			if (m_Slots[i].m_Outgoing.empty())

+			{

+				// Nothing ready yet

+				continue;

+			}

+		}  // if (outgoing data is empty)

+		

+		int Sent = m_Slots[i].m_Socket->Send(m_Slots[i].m_Outgoing.data(), m_Slots[i].m_Outgoing.size());

+		if (Sent < 0)

+		{

+			LOGWARNING("Error while writing to client \"%s\", disconnecting", m_Slots[i].m_Socket->GetIPString().c_str());

+			m_Slots[i].m_Socket->CloseSocket();

+			m_Slots[i].m_Client->SocketClosed();

+			return;

+		}

+		m_Slots[i].m_Outgoing.erase(0, Sent);

+	}

+}

+

+

+

+

diff --git a/source/cSocketThreads.h b/source/cSocketThreads.h
new file mode 100644
index 000000000..b43d693ba
--- /dev/null
+++ b/source/cSocketThreads.h
@@ -0,0 +1,147 @@
+

+// cSocketThreads.h

+

+// Interfaces to the cSocketThreads class representing the heart of MCS's client networking.

+// This object takes care of network communication, groups sockets into threads and uses as little threads as possible for full read / write support

+// For more detail, see http://forum.mc-server.org/showthread.php?tid=327

+

+

+

+

+

+/// How many clients should one thread handle? (must be less than FD_SETSIZE - 1 for your platform)

+#define MAX_SLOTS 1

+

+

+

+

+

+#pragma once

+#ifndef CSOCKETTHREADS_H_INCLUDED

+#define CSOCKETTHREADS_H_INCLUDED

+

+#include "packets/cPacket.h"

+#include "cIsThread.h"

+

+

+

+

+// Check MAX_SLOTS:

+#if MAX_SLOTS >= FD_SETSIZE

+	#error "MAX_SLOTS must be less than FD_SETSIZE - 1 for your platform! (otherwise select() won't work)"

+#endif

+

+

+

+

+

+// fwd:

+class cSocket;

+class cClientHandle;

+

+

+

+

+

+class cSocketThreads

+{

+public:

+

+	// Clients of cSocketThreads must implement this interface to be able to communicate

+	class cCallback

+	{

+	public:

+		/// Called when data is received from the remote party

+		virtual void DataReceived(const char * a_Data, int a_Size) = 0;

+		

+		/// Called when data can be sent to remote party; the function is supposed to append outgoing data to a_Data

+		virtual void GetOutgoingData(AString & a_Data) = 0;

+		

+		/// Called when the socket has been closed for any reason

+		virtual void SocketClosed(void) = 0;

+	} ;

+

+	

+	cSocketThreads(void);

+	~cSocketThreads();

+	

+	/// Add a (socket, client) pair for processing, data from a_Socket is to be sent to a_Client

+	void AddClient(cSocket * a_Socket, cCallback * a_Client);

+	

+	/// Remove the socket (and associated client) from processing

+	void RemoveClient(cSocket * a_Socket);

+	

+	/// Remove the associated socket and the client from processing

+	void RemoveClient(cCallback * a_Client);

+	

+	/// Notify the thread responsible for a_Client that the client has something to write

+	void NotifyWrite(cCallback * a_Client);

+

+private:

+

+	class cSocketThread :

+		public cIsThread

+	{

+		typedef cIsThread super;

+		

+	public:

+	

+		cSocketThread(cSocketThreads * a_Parent);

+		

+		// All these methods assume parent's m_CS is locked

+		bool HasEmptySlot(void) const {return m_NumSlots < MAX_SLOTS; }

+		bool IsEmpty     (void) const {return m_NumSlots == 0; }

+

+		void AddClient   (cSocket *   a_Socket, cCallback * a_Client);

+		bool RemoveClient(cCallback * a_Client);  // Returns true if removed, false if not found

+		bool RemoveSocket(cSocket *   a_Socket);  // Returns true if removed, false if not found

+		bool HasClient   (cCallback * a_Client) const;

+		bool HasSocket   (cSocket *   a_Socket) const;

+		bool NotifyWrite (cCallback * a_Client);  // Returns true if client handled by this thread

+		

+		bool Start(void);  // Hide the cIsThread's Start method, we need to provide our own startup to create the control socket

+		

+	private:

+	

+		cSocketThreads * m_Parent;

+	

+		// Two ends of the control socket, the first is select()-ed, the second is written to for notifications

+		cSocket m_ControlSocket1;

+		cSocket m_ControlSocket2;

+		

+		// Socket-client-packetqueues triplets.

+		// Manipulation with these assumes that the parent's m_CS is locked

+		struct sSlot

+		{

+			cSocket *   m_Socket;

+			cCallback * m_Client;

+			AString     m_Outgoing;  // If sending writes only partial data, the rest is stored here for another send

+		} ;

+		sSlot m_Slots[MAX_SLOTS];

+		int   m_NumSlots;  // Number of slots actually used

+		

+		virtual void Execute(void) override;

+		

+		void AddOrUpdatePacket(int a_Slot, cPacket * a_Packet);  // Adds the packet to the specified slot, or updates an existing packet in that queue (EntityMoveLook filtering)

+		

+		void PrepareSet     (fd_set * a_Set, cSocket::xSocket & a_Highest);  // Puts all sockets into the set, along with m_ControlSocket1

+		void ReadFromSockets(fd_set * a_Read);  // Reads from sockets indicated in a_Read

+		void WriteToSockets (fd_set * a_Write);  // Writes to sockets indicated in a_Write

+	} ;

+	

+	typedef std::list<cSocketThread *> cSocketThreadList;

+	

+	

+	cCriticalSection  m_CS;

+	cSocketThreadList m_Threads;

+} ;

+

+

+

+

+

+#endif  // CSOCKETTHREADS_H_INCLUDED

+

+

+

+