diff options
Diffstat (limited to '')
| -rw-r--r-- | source/cSocketThreads.cpp | 472 |
1 files changed, 472 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);
+ }
+}
+
+
+
+
|