Tremendously improving the speed of the network send code (ticket #52). The code has also been simplified, and uses some kind of double buffering as in graphics programming. There are now a lot less memory allocations and a lot less copying. However, memory usage is also increased.

src/net/common/senderhelper.cpp

@@ -20,12 +20,12 @@
 #include <boost/asio.hpp>
 #include <boost/bind.hpp>
 #include <boost/enable_shared_from_this.hpp>
+#include <boost/swap.hpp>
 
 #include <net/senderhelper.h>
 #include <net/sendercallback.h>
 #include <net/socket_helper.h>
 #include <net/socket_msg.h>
-#include <net/encodedpacket.h>
 #include <core/loghelper.h>
 #include <cstring>
 #include <cassert>
@@ -34,80 +34,103 @@ using namespace std;
 using boost::asio::ip::tcp;
 
 
-#define SEND_ERROR_TIMEOUT_MSEC				20000
+#define SEND_BUF_FIRST_ALLOC_CHUNKSIZE		4096
-#define SEND_TIMEOUT_MSEC					10
+#define MAX_SEND_BUF_SIZE					SEND_BUF_FIRST_ALLOC_CHUNKSIZE * 1000
-#define SEND_QUEUE_SIZE						10000000
-#define SEND_LOG_INTERVAL_SEC				60
 
 
-typedef std::list<boost::shared_ptr<EncodedPacket> > SendDataList;
-
 class SendDataManager : public boost::enable_shared_from_this<SendDataManager>
 {
 public:
 	SendDataManager()
-		: sendBuf(NULL), writeInProgress(false) {
+		: sendBuf(NULL), curWriteBuf(NULL), sendBufAllocated(0), sendBufUsed(0),
+		  curWriteBufAllocated(0), curWriteBufUsed(0)
+	{
 	}
 
-	~SendDataManager() {
+	~SendDataManager()
-		delete[] sendBuf;
+	{
+		free(sendBuf);
+		free(curWriteBuf);
+	}
+
+	inline size_t GetSendBufLeft() const
+	{
+		int bytesLeft = sendBufAllocated - sendBufUsed;
+		return bytesLeft < 0 ? (size_t)0 : (size_t)bytesLeft;
+	}
+
+	inline size_t GetAllocated() const
+	{
+		return sendBufAllocated;
+	}
+
+	inline bool ReallocSendBuf()
+	{
+		bool retVal = false;
+		size_t allocAmount = sendBufAllocated * 2;
+		if (0 == allocAmount) {
+			allocAmount = (size_t)SEND_BUF_FIRST_ALLOC_CHUNKSIZE;
+		}
+		char *tempBuf = (char *)realloc(sendBuf, allocAmount);
+		if (tempBuf) {
+			sendBuf = tempBuf;
+			sendBufAllocated = allocAmount;
+			retVal = true;
+		}
+		return retVal;
+	}
+
+	inline void AppendToSendBufWithoutCheck(const char *data, size_t size)
+	{
+		memcpy(sendBuf + sendBufUsed, data, size);
+		sendBufUsed += size;
 	}
 
 	void HandleWrite(boost::shared_ptr<boost::asio::ip::tcp::socket> socket, const boost::system::error_code &error);
 
-	void AsyncSendNextPacket(boost::shared_ptr<boost::asio::ip::tcp::socket> socket, bool handlerMode = false);
+	void AsyncSendNextPacket(boost::shared_ptr<boost::asio::ip::tcp::socket> socket);
 
 	mutable boost::mutex dataMutex;
-	SendDataList list;
+
+private:
 	char *sendBuf;
-	bool writeInProgress;
+	char *curWriteBuf;
+	size_t sendBufAllocated;
+	size_t sendBufUsed;
+	size_t curWriteBufAllocated;
+	size_t curWriteBufUsed;
 };
 
 
 void
 SendDataManager::HandleWrite(boost::shared_ptr<boost::asio::ip::tcp::socket> socket, const boost::system::error_code &error)
 {
-	if (!error)
+	if (!error) {
-		AsyncSendNextPacket(socket, true);
+		// Successfully sent the data.
+		curWriteBufUsed = 0;
+		// Send more data, if available.
+		AsyncSendNextPacket(socket);
+	}
 }
 
 void
-SendDataManager::AsyncSendNextPacket(boost::shared_ptr<boost::asio::ip::tcp::socket> socket, bool handlerMode)
+SendDataManager::AsyncSendNextPacket(boost::shared_ptr<boost::asio::ip::tcp::socket> socket)
 {
 	boost::mutex::scoped_lock lock(dataMutex);
-	if (!writeInProgress || handlerMode) {
+	if (!curWriteBufUsed) {
-		delete[] sendBuf;
+		// Swap buffers and send data.
-		sendBuf = NULL;
+		boost::swap(curWriteBuf, sendBuf);
-
+		boost::swap(curWriteBufAllocated, sendBufAllocated);
-		unsigned bufPos = 0;
+		boost::swap(curWriteBufUsed, sendBufUsed);
-		unsigned bufSize = 0;
+		if (curWriteBufUsed) {
-		// Count required bytes.
-		SendDataList::iterator i = list.begin();
-		SendDataList::iterator end = list.end();
-		while (i != end) {
-			bufSize += (*i)->GetSize();
-			++i;
-		}
-		if (bufSize) {
-			sendBuf = new char[bufSize];
-			i = list.begin();
-			end = list.end();
-			while (i != end) {
-				memcpy(sendBuf + bufPos, (*i)->GetData(), (*i)->GetSize());
-				bufPos += (*i)->GetSize();
-				++i;
-			}
-			list.clear();
 			boost::asio::async_write(
 				*socket,
-				boost::asio::buffer(sendBuf, bufSize),
+				boost::asio::buffer(curWriteBuf, curWriteBufUsed),
 				boost::bind(&SendDataManager::HandleWrite,
 							shared_from_this(),
 							socket,
 							boost::asio::placeholders::error));
-			writeInProgress = true;
+		}
-		} else
-			writeInProgress = false;
 	}
 }
 
@@ -136,7 +159,7 @@ SenderHelper::Send(boost::shared_ptr<SessionData> session, boost::shared_ptr<Net
 		{
 			// Second: Add packet to specific queue.
 			boost::mutex::scoped_lock lock(tmpManager->dataMutex);
-			if (tmpManager->list.size() < SEND_QUEUE_SIZE) {
+			if (tmpManager->GetAllocated() <= MAX_SEND_BUF_SIZE) {
 				InternalStorePacket(*tmpManager, packet);
 			}
 		}
@@ -163,7 +186,7 @@ SenderHelper::Send(boost::shared_ptr<SessionData> session, const NetPacketList &
 		{
 			// Second: Add packets to specific queue.
 			boost::mutex::scoped_lock lock(tmpManager->dataMutex);
-			if (tmpManager->list.size() + packetList.size() <= SEND_QUEUE_SIZE) {
+			if (tmpManager->GetAllocated() <= MAX_SEND_BUF_SIZE) {
 				NetPacketList::const_iterator i = packetList.begin();
 				NetPacketList::const_iterator end = packetList.end();
 				while (i != end) {
@@ -190,15 +213,27 @@ SenderHelper::SignalSessionTerminated(unsigned sessionId)
 		m_sendQueueMap.erase(pos);
 }
 
+static int der_encode_to_buffer_cb(const void *data, size_t size, void *arg) {
+	SendDataManager *m = (SendDataManager *)arg;
+
+	// Realloc buffer if necessary.
+	while (m->GetSendBufLeft() < size) {
+		if (!m->ReallocSendBuf()) {
+			return -1;
+		}
+	}
+
+	m->AppendToSendBufWithoutCheck((const char*)data, size);
+
+	return 0;
+}
+
 void
 SenderHelper::InternalStorePacket(SendDataManager &tmpManager, boost::shared_ptr<NetPacket> packet)
 {
-	unsigned char buf[MAX_PACKET_SIZE];
+	asn_enc_rval_t e = der_encode(&asn_DEF_PokerTHMessage, packet->GetMsg(), der_encode_to_buffer_cb, &tmpManager);
-	asn_enc_rval_t e = der_encode_to_buffer(&asn_DEF_PokerTHMessage, packet->GetMsg(), buf, MAX_PACKET_SIZE);
 	//cerr << "OUT:" << endl << packet->ToString() << endl;
 	if (e.encoded == -1)
 		LOG_ERROR("Failed to encode NetPacket: " << packet->GetMsg()->present);
-	else
-		tmpManager.list.push_back(boost::shared_ptr<EncodedPacket>(new EncodedPacket(buf, e.encoded)));
 }