More work on sending/receiving avatars. Introduces low priority send queue, so that the server send thread is not stalled when sending loads of avatars. Not functional yet.

src/core/avatarmanager.h

@@ -21,16 +21,16 @@
 #ifndef _AVATARMANAGER_H_
 #define _AVATARMANAGER_H_
 
+#include <playerdata.h>
+#include <net/netpacket.h>
 #include <core/crypthelper.h>
 #include <map>
 
 #include <boost/shared_ptr.hpp>
-#include <boost/function.hpp>
 
 #define MAX_AVATAR_FILE_SIZE	30720
 
 struct AvatarFileState;
-class NetPacket;
 
 class AvatarManager
 {
@@ -41,14 +41,14 @@ public:
 
 	bool Init(const std::string &dataDir, const std::string &cacheDir);
 
-	boost::shared_ptr<AvatarFileState> OpenAvatarFileForChunkRead(const std::string &fileName, unsigned &outFileSize);
+	boost::shared_ptr<AvatarFileState> OpenAvatarFileForChunkRead(const std::string &fileName, unsigned &outFileSize, AvatarFileType &outFileType);
 	unsigned ChunkReadAvatarFile(boost::shared_ptr<AvatarFileState> fileState, unsigned char *data, unsigned chunkSize);
 
-	bool SendAvatarFile(const std::string &fileName, unsigned requestId, boost::function<void (boost::shared_ptr<NetPacket>)> sender);
+	bool AvatarFileToNetPackets(const std::string &fileName, unsigned requestId, NetPacketList &packets);
 
 	bool GetHashForAvatar(const std::string &fileName, MD5Buf &md5buf);
 	bool GetAvatarFileName(const MD5Buf &md5buf, std::string &fileName) const;
-	bool StoreAvatarInCache(const MD5Buf &md5buf, const std::string &fileExtension, const unsigned char *data, unsigned size);
+	bool StoreAvatarInCache(const MD5Buf &md5buf, AvatarFileType avatarFileType, const unsigned char *data, unsigned size);
 
 protected:
 	typedef std::map<MD5Buf, std::string> AvatarMap;

src/core/common/avatarmanager.cpp

@@ -18,12 +18,19 @@
  ***************************************************************************/
 
 #include "avatarmanager.h"
-#include <net/netpacket.h>
 
 #include <boost/filesystem.hpp>
 #include <openssl/md5.h>
 
 #include <fstream>
+#include <cstring>
+
+// Not using boost::algorithm here because of STL issues.
+#ifdef _MSC_VER
+#define STRCASECMP _stricmp
+#else
+#define STRCASECMP strcasecmp
+#endif
 
 using namespace std;
 using namespace boost::filesystem;
@@ -59,12 +66,21 @@ AvatarManager::Init(const std::string &dataDir, const std::string &cacheDir)
 }
 
 boost::shared_ptr<AvatarFileState>
-AvatarManager::OpenAvatarFileForChunkRead(const std::string &fileName, unsigned &outFileSize)
+AvatarManager::OpenAvatarFileForChunkRead(const std::string &fileName, unsigned &outFileSize, AvatarFileType &outFileType)
 {
 	outFileSize = 0;
+	outFileType = AVATAR_FILE_TYPE_UNKNOWN;
 	boost::shared_ptr<AvatarFileState> retVal;
 	try
 	{
+		path filePath(fileName);
+		string ext(extension(filePath));
+		if (STRCASECMP(ext.c_str(), ".png") == 0)
+			outFileType = AVATAR_FILE_TYPE_PNG;
+		else if (STRCASECMP(ext.c_str(), ".jpg") == 0 || STRCASECMP(ext.c_str(), ".jpeg") == 0)
+			outFileType = AVATAR_FILE_TYPE_JPG;
+		else if (STRCASECMP(ext.c_str(), ".gif") == 0)
+			outFileType = AVATAR_FILE_TYPE_GIF;
 		boost::shared_ptr<AvatarFileState> fileState(new AvatarFileState);
 		fileState->inputStream.open(fileName.c_str(), ios_base::in | ios_base::binary);
 		if (!fileState->inputStream.fail())
@@ -108,20 +124,21 @@ AvatarManager::ChunkReadAvatarFile(boost::shared_ptr<AvatarFileState> fileState,
 }
 
 bool
-AvatarManager::SendAvatarFile(const string &fileName, unsigned requestId, boost::function<void (boost::shared_ptr<NetPacket>)> sender)
+AvatarManager::AvatarFileToNetPackets(const string &fileName, unsigned requestId, NetPacketList &packets)
 {
 	bool retVal = false;
 	unsigned fileSize;
-	boost::shared_ptr<AvatarFileState> tmpState = OpenAvatarFileForChunkRead(fileName, fileSize);
+	AvatarFileType fileType;
-	if (tmpState.get() && fileSize)
+	boost::shared_ptr<AvatarFileState> tmpState = OpenAvatarFileForChunkRead(fileName, fileSize, fileType);
+	if (tmpState.get() && fileSize && fileType != AVATAR_FILE_TYPE_UNKNOWN)
 	{
 		boost::shared_ptr<NetPacket> avatarHeader(new NetPacketAvatarHeader);
 		NetPacketAvatarHeader::Data avatarHeaderData;
 		avatarHeaderData.requestId = requestId;
 		avatarHeaderData.avatarFileSize = fileSize;
-		avatarHeaderData.avatarFileType = AVATAR_TYPE_PNG; // TODO
+		avatarHeaderData.avatarFileType = fileType;
 		static_cast<NetPacketAvatarHeader *>(avatarHeader.get())->SetData(avatarHeaderData);
-		sender(avatarHeader);
+		packets.push_back(avatarHeader);
 
 		unsigned numBytes = 0;
 		unsigned totalBytesRead = 0;
@@ -137,7 +154,7 @@ AvatarManager::SendAvatarFile(const string &fileName, unsigned requestId, boost:
 				avatarFileData.requestId = requestId;
 				totalBytesRead += numBytes;
 				static_cast<NetPacketAvatarFile *>(avatarFile.get())->SetData(avatarFileData);
-				sender(avatarFile);
+				packets.push_back(avatarFile);
 			}
 		} while (numBytes);
 		// TODO error handling if numBytes != totalBytesRead
@@ -145,7 +162,7 @@ AvatarManager::SendAvatarFile(const string &fileName, unsigned requestId, boost:
 		NetPacketAvatarEnd::Data avatarEndData;
 		avatarEndData.requestId = requestId;
 		static_cast<NetPacketAvatarEnd *>(avatarEnd.get())->SetData(avatarEndData);
-		sender(avatarEnd);
+		packets.push_back(avatarEnd);
 		retVal = true;
 	}
 	// else TODO error handling
@@ -197,13 +214,26 @@ AvatarManager::GetAvatarFileName(const MD5Buf &md5buf, std::string &fileName) co
 }
 
 bool
-AvatarManager::StoreAvatarInCache(const MD5Buf &md5buf, const std::string &fileExtension, const unsigned char *data, unsigned size)
+AvatarManager::StoreAvatarInCache(const MD5Buf &md5buf, AvatarFileType avatarFileType, const unsigned char *data, unsigned size)
 {
 	bool retVal = false;
 	try
 	{
+		string ext;
+		switch (avatarFileType)
+		{
+			case AVATAR_FILE_TYPE_PNG:
+				ext = ".png";
+				break;
+			case AVATAR_FILE_TYPE_JPG:
+				ext = ".jpg";
+				break;
+			case AVATAR_FILE_TYPE_GIF:
+				ext = ".gif";
+				break;
+		}
 		path tmpPath(m_cacheDir);
-		tmpPath /= (md5buf.ToString() + "." + fileExtension);
+		tmpPath /= (md5buf.ToString() + ext);
 		string fileName(tmpPath.file_string());
 		ofstream o(fileName.c_str(), ios_base::out | ios_base::binary);
 		o.write((const char *)data, size);

src/net/common/clientstate.cpp

@@ -32,6 +32,8 @@
 #include <game.h>
 #include <playerinterface.h>
 
+#include <boost/bind.hpp>
+
 #include <sstream>
 
 using namespace std;
@@ -529,10 +531,18 @@ ClientStateWaitSession::InternalProcess(ClientThread &client, boost::shared_ptr<
 	}
 	else if (packet->ToNetPacketRetrieveAvatar())
 	{
+		// Before letting us join the lobby, the server requests our avatar.
 		NetPacketRetrieveAvatar::Data retrieveAvatarData;
 		packet->ToNetPacketRetrieveAvatar()->GetData(retrieveAvatarData);
 
-		// Before letting us join the lobby, the server requests our avatar.
+		NetPacketList tmpList;
+		if (client.GetAvatarManager().AvatarFileToNetPackets(
+			client.GetContext().GetAvatarFile(),
+			retrieveAvatarData.requestId,
+			tmpList))
+		{
+			client.GetSender().SendLowPrio(client.GetContext().GetSocket(), tmpList);
+		}
 	}
 
 	return retVal;

src/net/common/senderthread.cpp

@@ -22,6 +22,8 @@
 #include <net/socket_msg.h>
 #include <cstring>
 
+#include <boost/bind.hpp>
+
 using namespace std;
 
 
@@ -40,12 +42,64 @@ SenderThread::Send(SOCKET sock, boost::shared_ptr<NetPacket> packet)
 	if (packet.get() && IS_VALID_SOCKET(sock))
 	{
 		boost::mutex::scoped_lock lock(m_outBufMutex);
-		if (m_outBuf.size() < SEND_QUEUE_SIZE) // Queue is limited in size.
+		InternalStore(m_outBuf, SEND_QUEUE_SIZE, sock, packet);
-			m_outBuf.push_back(std::make_pair(packet, sock));
-		// TODO: Throw exception if failed.
 	}
 }
 
+void
+SenderThread::Send(SOCKET sock, const NetPacketList &packetList)
+{
+	if (!packetList.empty() && IS_VALID_SOCKET(sock))
+	{
+		boost::mutex::scoped_lock lock(m_outBufMutex);
+		InternalStore(m_outBuf, SEND_QUEUE_SIZE, sock, packetList);
+	}
+}
+
+void
+SenderThread::SendLowPrio(SOCKET sock, boost::shared_ptr<NetPacket> packet)
+{
+	if (packet.get() && IS_VALID_SOCKET(sock))
+	{
+		boost::mutex::scoped_lock lock(m_lowPrioOutBufMutex);
+		InternalStore(m_lowPrioOutBuf, SEND_LOW_PRIO_QUEUE_SIZE, sock, packet);
+	}
+}
+
+void
+SenderThread::SendLowPrio(SOCKET sock, const NetPacketList &packetList)
+{
+	if (!packetList.empty() && IS_VALID_SOCKET(sock))
+	{
+		boost::mutex::scoped_lock lock(m_lowPrioOutBufMutex);
+		InternalStore(m_lowPrioOutBuf, SEND_LOW_PRIO_QUEUE_SIZE, sock, packetList);
+	}
+}
+
+void
+SenderThread::InternalStore(SendDataDeque &sendQueue, unsigned maxQueueSize, SOCKET sock, boost::shared_ptr<NetPacket> packet)
+{
+	if (sendQueue.size() < maxQueueSize) // Queue is limited in size.
+		sendQueue.push_back(std::make_pair(packet, sock));
+	// TODO: Throw exception if failed.
+}
+
+void
+SenderThread::InternalStore(SendDataDeque &sendQueue, unsigned maxQueueSize, SOCKET sock, const NetPacketList &packetList)
+{
+	if (sendQueue.size() + packetList.size() < maxQueueSize)
+	{
+		NetPacketList::const_iterator i = packetList.begin();
+		NetPacketList::const_iterator end = packetList.end();
+		while (i != end)
+		{
+			sendQueue.push_back(std::make_pair(*i, sock));
+			++i;
+		}
+	}
+	// TODO: Throw exception if failed.
+}
+
 void
 SenderThread::Main()
 {
@@ -57,6 +111,7 @@ SenderThread::Main()
 		if (!m_tmpOutBufSize)
 		{
 			SendData tmpData;
+			// Check main queue first.
 			{
 				boost::mutex::scoped_lock lock(m_outBufMutex);
 				if (!m_outBuf.empty())
@@ -66,6 +121,17 @@ SenderThread::Main()
 				}
 			}
 
+			// Check low prio queue only if there is nothing in the main queue.
+			if (!tmpData.first.get())
+			{
+				boost::mutex::scoped_lock lock(m_lowPrioOutBufMutex);
+				if (!m_lowPrioOutBuf.empty())
+				{
+					tmpData = m_lowPrioOutBuf.front();
+					m_lowPrioOutBuf.pop_front();
+				}
+			}
+
 			if (tmpData.first.get())
 			{
 				if (IS_VALID_SOCKET(tmpData.second))

src/net/netpacket.h

@@ -28,6 +28,7 @@
 #include <core/crypthelper.h>
 
 #include <vector>
+#include <list>
 
 #define NET_VERSION_MAJOR			2
 #define NET_VERSION_MINOR			0
@@ -158,6 +159,8 @@ private:
 	const u_int16_t m_maxSize;
 };
 
+typedef std::list<boost::shared_ptr<NetPacket> > NetPacketList;
+
 class NetPacketInit : public NetPacket
 {
 public:

src/net/senderthread.h

@@ -31,7 +31,8 @@
 
 #define SENDER_THREAD_TERMINATE_TIMEOUT		200
 #define SEND_TIMEOUT_MSEC					10
-#define SEND_QUEUE_SIZE						500
+#define SEND_QUEUE_SIZE						1000
+#define SEND_LOW_PRIO_QUEUE_SIZE			50000
 
 class SenderThread : public Thread
 {
@@ -40,20 +41,31 @@ public:
 	virtual ~SenderThread();
 
 	void Send(SOCKET sock, boost::shared_ptr<NetPacket> packet);
+	void Send(SOCKET sock, const NetPacketList &packetList);
+
+	void SendLowPrio(SOCKET sock, boost::shared_ptr<NetPacket> packet);
+	void SendLowPrio(SOCKET sock, const NetPacketList &packetList);
 
 protected:
+	typedef std::pair<boost::shared_ptr<NetPacket>, SOCKET> SendData;
+	typedef std::deque<SendData> SendDataDeque;
 
 	// Main function of the thread.
 	virtual void Main();
 
+	void InternalStore(SendDataDeque &sendQueue, unsigned maxQueueSize, SOCKET sock, boost::shared_ptr<NetPacket> packet);
+	void InternalStore(SendDataDeque &sendQueue, unsigned maxQueueSize, SOCKET sock, const NetPacketList &packetList);
+
 private:
 
 	SOCKET m_curSocket;
 
-	typedef std::pair<boost::shared_ptr<NetPacket>, SOCKET> SendData;
 	std::deque<SendData> m_outBuf;
 	mutable boost::mutex m_outBufMutex;
 
+	std::deque<SendData> m_lowPrioOutBuf;
+	mutable boost::mutex m_lowPrioOutBufMutex;
+
 	char m_tmpOutBuf[MAX_PACKET_SIZE];
 	unsigned m_tmpOutBufSize;
 

src/playerdata.h

@@ -44,9 +44,10 @@ enum PlayerRights
 
 enum AvatarFileType
 {
-	AVATAR_TYPE_PNG = 1,
+	AVATAR_FILE_TYPE_UNKNOWN = 0,
-	AVATAR_TYPE_JPG,
+	AVATAR_FILE_TYPE_PNG,
-	AVATAR_TYPE_GIF
+	AVATAR_FILE_TYPE_JPG,
+	AVATAR_FILE_TYPE_GIF
 };
 
 struct PlayerInfo