Bug fixes and implementation cleanup

common/eq_stream_intf.h

@@ -29,7 +29,6 @@ struct EQStreamManagerInterfaceOptions
 
 	EQStreamManagerInterfaceOptions(int port, bool encoded, bool compressed) {
 		opcode_size = 2;
-		track_opcode_stats = false;
 
 		//World seems to support both compression and xor zone supports one or the others.
 		//Enforce one or the other in the convienence construct
@@ -47,7 +46,6 @@ struct EQStreamManagerInterfaceOptions
 	}
 
 	int opcode_size;
-	bool track_opcode_stats;
 	EQ::Net::DaybreakConnectionManagerOptions daybreak_options;
 	EQ::EventLoop *loop;
 };

common/net/eqstream.cpp

@@ -65,9 +65,7 @@ void EQ::Net::EQStream::QueuePacket(const EQApplicationPacket *p, bool ack_req)
 			opcode = p->GetOpcodeBypass();
 		}
 		else {
-			if (m_owner->GetOptions().track_opcode_stats) {
+			m_packet_sent_count[p->GetOpcode()]++; //Wont bother with bypass tracking of these since those are rare for testing anyway
-				m_packet_sent_count[p->GetOpcode()]++; //Wont bother with bypass tracking of these since those are rare for testing anyway
-			}
 			opcode = (*m_opcode_manager)->EmuToEQ(p->GetOpcode());
 		}
 
@@ -117,9 +115,7 @@ EQApplicationPacket *EQ::Net::EQStream::PopPacket() {
 		}
 
 		EmuOpcode emu_op = (*m_opcode_manager)->EQToEmu(opcode);
-		if (m_owner->GetOptions().track_opcode_stats) {
+		m_packet_recv_count[emu_op]++;
-			m_packet_recv_count[emu_op]++;
-		}
 
 		EQApplicationPacket *ret = new EQApplicationPacket(emu_op, (unsigned char*)p->Data() + m_owner->GetOptions().opcode_size, p->Length() - m_owner->GetOptions().opcode_size);
 		ret->SetProtocolOpcode(opcode);

common/net/eqstream_concurrent.cpp

@@ -4,6 +4,8 @@
 #include "../event/timer.h"
 #include "../string_util.h"
 #include "../opcodemgr.h"
+#include "../eqemu_logsys.h"
+#include "../eqemu_logsys_fmt.h"
 #include "daybreak_connection.h"
 #include <thread>
 #include <concurrentqueue.h>
@@ -56,7 +58,7 @@ EQ::Net::ConcurrentEQStreamManager::~ConcurrentEQStreamManager()
 
 	while (_impl->foreground_queue.try_dequeue(eqs_msg)) {
 		if (eqs_msg.type == PacketRecv) {
-			ConcurrentEQStreamPacketRecvMessage *eqs_msg_in = (ConcurrentEQStreamPacketRecvMessage*)&eqs_msg;
+			ceqs_packet_recv_msg_t *eqs_msg_in = (ceqs_packet_recv_msg_t*)&eqs_msg;
 
 			delete eqs_msg_in->packet;
 		}
@@ -92,8 +94,8 @@ void EQ::Net::ConcurrentEQStreamManager::_BackgroundThread() {
 
 	ceqs_msg_t eqs_msg;
 	while (_impl->background_queue.try_dequeue(eqs_msg)) {
-		if (eqs_msg.type == PacketRecv) {
+		if (eqs_msg.type == QueuePacket) {
-			ConcurrentEQStreamPacketRecvMessage *eqs_msg_in = (ConcurrentEQStreamPacketRecvMessage*)&eqs_msg;
+			ceqs_queue_packet_msg_t *eqs_msg_in = (ceqs_queue_packet_msg_t*)&eqs_msg;
 			delete eqs_msg_in->packet;
 		}
 	}
@@ -118,22 +120,21 @@ void EQ::Net::ConcurrentEQStreamManager::_BackgroundUpdateStatsTimer(EQ::Timer *
 
 	for (auto &c : _impl->connections) {
 		auto &connection = c.second;
-		auto msg = (ceqs_update_daybreak_stats_msg_t*)&msgs[i];
+		auto msg = (ceqs_update_stats_msg_t*)&msgs[i];
 
-		msg->type = ceqs_msg_type::UpdateDaybreakStats;
+		msg->type = ceqs_msg_type::UpdateStats;
 		msg->stream_id = connection->GetId();
 		msg->stats = connection->GetStats();
 		i++;
 
-		printf("Sending stats to client %u\n", connection->GetId());
 		if (i >= 16) {
-			_impl->background_queue.enqueue_bulk(msgs, 16);
+			_impl->foreground_queue.enqueue_bulk(msgs, 16);
 			i = 0;
 		}
 	}
 
 	if (i > 0) {
-		_impl->background_queue.enqueue_bulk(msgs, i);
+		_impl->foreground_queue.enqueue_bulk(msgs, i);
 	}
 }
 
@@ -143,8 +144,7 @@ void EQ::Net::ConcurrentEQStreamManager::_ProcessBackgroundMessage(const ceqs_ms
 	switch (msg.type) {
 	case QueuePacket:
 	{
-		ConcurrentEQStreamQueuePacketMessage *msg_in = (ConcurrentEQStreamQueuePacketMessage*)&msg;
+		ceqs_queue_packet_msg_t *msg_in = (ceqs_queue_packet_msg_t*)&msg;
-		printf("(background) Packet Queue for %u with %u bytes with ack: %s\n", msg_in->stream_id, msg_in->packet->Length(), msg_in->ack_req ? "true" : "false");
 
 		auto iter = _impl->connections.find(msg_in->stream_id);
 		if (iter != _impl->connections.end()) {
@@ -161,7 +161,7 @@ void EQ::Net::ConcurrentEQStreamManager::_ProcessBackgroundMessage(const ceqs_ms
 	}
 	case CloseConnection:
 	{
-		ConcurrentEQStreamCloseConnectionMessage *msg_in = (ConcurrentEQStreamCloseConnectionMessage*)&msg;
+		ceqs_close_connection_msg_t *msg_in = (ceqs_close_connection_msg_t*)&msg;
 		auto iter = _impl->connections.find(msg_in->stream_id);
 		if (iter != _impl->connections.end()) {
 			iter->second->Close();
@@ -170,7 +170,7 @@ void EQ::Net::ConcurrentEQStreamManager::_ProcessBackgroundMessage(const ceqs_ms
 	}
 	case ResetStats:
 	{
-		ConcurrentEQStreamResetStatsMessage *msg_in = (ConcurrentEQStreamResetStatsMessage*)&msg;
+		ceqs_reset_stats_msg_t *msg_in = (ceqs_reset_stats_msg_t*)&msg;
 		auto iter = _impl->connections.find(msg_in->stream_id);
 		if (iter != _impl->connections.end()) {
 			iter->second->ResetStats();
@@ -178,7 +178,7 @@ void EQ::Net::ConcurrentEQStreamManager::_ProcessBackgroundMessage(const ceqs_ms
 		break;
 	}
 	default:
-		printf("(background) New message with unhandled type %u\n", (int)msg.type);
+		break;
 	}
 }
 
@@ -200,8 +200,7 @@ void EQ::Net::ConcurrentEQStreamManager::_ProcessForegroundMessage(const ceqs_ms
 	switch (msg.type) {
 	case NewConnection:
 	{
-		ConcurrentEQStreamNewConnectionMessage *msg_in = (ConcurrentEQStreamNewConnectionMessage*)&msg;
+		ceqs_new_connection_msg_t *msg_in = (ceqs_new_connection_msg_t*)&msg;
-		printf("(foreground) New connection from %s:%u with id: %u\n", msg_in->endpoint, msg_in->remote_port, msg_in->stream_id);
 
 		std::shared_ptr<ConcurrentEQStream> stream(new ConcurrentEQStream(this, 
 			msg_in->stream_id, 
@@ -217,9 +216,7 @@ void EQ::Net::ConcurrentEQStreamManager::_ProcessForegroundMessage(const ceqs_ms
 	}
 	case ConnectionStateChange:
 	{
-		ConcurrentEQStreamConnectionStateChangeMessage *msg_in = (ConcurrentEQStreamConnectionStateChangeMessage*)&msg;
+		ceqs_connection_state_change_msg_t *msg_in = (ceqs_connection_state_change_msg_t*)&msg;
-		printf("(foreground) Connection State Change for %u, was %u now is %u\n", msg_in->stream_id, msg_in->from, msg_in->to);
-
 
 		auto iter = _impl->streams.find(msg_in->stream_id);
 		if (iter != _impl->streams.end()) {
@@ -233,8 +230,7 @@ void EQ::Net::ConcurrentEQStreamManager::_ProcessForegroundMessage(const ceqs_ms
 	}
 	case PacketRecv:
 	{
-		ConcurrentEQStreamPacketRecvMessage *msg_in = (ConcurrentEQStreamPacketRecvMessage*)&msg;
+		ceqs_packet_recv_msg_t *msg_in = (ceqs_packet_recv_msg_t*)&msg;
-		printf("(foreground) Packet Recv for %u with %u bytes\n", msg_in->stream_id, msg_in->packet->Length());
 		std::unique_ptr<EQ::Net::Packet> p(msg_in->packet);
 
 		auto iter = _impl->streams.find(msg_in->stream_id);
@@ -243,9 +239,9 @@ void EQ::Net::ConcurrentEQStreamManager::_ProcessForegroundMessage(const ceqs_ms
 		}
 		break;
 	}
-	case UpdateDaybreakStats:
+	case UpdateStats:
 	{
-		ceqs_update_daybreak_stats_msg_t *msg_in = (ceqs_update_daybreak_stats_msg_t*)&msg;
+		ceqs_update_stats_msg_t *msg_in = (ceqs_update_stats_msg_t*)&msg;
 		auto iter = _impl->streams.find(msg_in->stream_id);
 		if (iter != _impl->streams.end()) {
 			iter->second->_UpdateStats(msg_in->stats);
@@ -283,7 +279,7 @@ void EQ::Net::ConcurrentEQStreamManager::OnConnectionStateChange(std::function<v
 void EQ::Net::ConcurrentEQStreamManager::DaybreakNewConnection(std::shared_ptr<DaybreakConnection> connection)
 {
 	_impl->connections.insert(std::make_pair(connection->GetId(), connection));
-	ConcurrentEQStreamNewConnectionMessage msg;
+	ceqs_new_connection_msg_t msg;
 	msg.type = ceqs_msg_type::NewConnection;
 	msg.stream_id = connection->GetId();
 	msg.remote_port = connection->RemotePort();
@@ -293,7 +289,6 @@ void EQ::Net::ConcurrentEQStreamManager::DaybreakNewConnection(std::shared_ptr<D
 
 	//Make sure the foreground gets this message
 	_PushToForegroundQueue((ceqs_msg_t*)&msg);
-	printf("(background) New connection from %s:%u with id: %u\n", connection->RemoteEndpoint().c_str(), connection->RemotePort(), connection->GetId());
 }
 
 //Called by background
@@ -306,7 +301,7 @@ void EQ::Net::ConcurrentEQStreamManager::DaybreakConnectionStateChange(std::shar
 		}
 	}
 
-	ConcurrentEQStreamConnectionStateChangeMessage msg;
+	ceqs_connection_state_change_msg_t msg;
 	msg.type = ceqs_msg_type::ConnectionStateChange;
 	msg.stream_id = connection->GetId();
 	msg.from = (int)from;
@@ -314,13 +309,12 @@ void EQ::Net::ConcurrentEQStreamManager::DaybreakConnectionStateChange(std::shar
 
 	//Make sure the foreground gets this message
 	_PushToForegroundQueue((ceqs_msg_t*)&msg);
-	printf("(background) Connection State Change for %u, was %u now is %u\n", connection->GetId(), (int)from, (int)to);
 }
 
 //Called by background
 void EQ::Net::ConcurrentEQStreamManager::DaybreakPacketRecv(std::shared_ptr<DaybreakConnection> connection, const Packet &p)
 {
-	ConcurrentEQStreamPacketRecvMessage msg;
+	ceqs_packet_recv_msg_t msg;
 	msg.type = ceqs_msg_type::PacketRecv;
 	msg.stream_id = connection->GetId();
 	msg.packet = new DynamicPacket();
@@ -328,7 +322,6 @@ void EQ::Net::ConcurrentEQStreamManager::DaybreakPacketRecv(std::shared_ptr<Dayb
 
 	//Make sure the foreground gets this message
 	_PushToForegroundQueue((ceqs_msg_t*)&msg);
-	printf("(background) Packet Recv for %u with %u bytes\n", connection->GetId(), p.Length());
 }
 
 struct EQ::Net::ConcurrentEQStream::Impl
@@ -342,6 +335,8 @@ struct EQ::Net::ConcurrentEQStream::Impl
 	std::deque<std::unique_ptr<EQ::Net::Packet>> packet_queue;
 	OpcodeManager **opcode_manager;
 	DaybreakConnectionStats stats;
+	std::unordered_map<EmuOpcode, int> packet_recv_count;
+	std::unordered_map<EmuOpcode, int> packet_sent_count;
 };
 
 //Called by foreground
@@ -376,9 +371,7 @@ void EQ::Net::ConcurrentEQStream::QueuePacket(const EQApplicationPacket *p, bool
 			opcode = p->GetOpcodeBypass();
 		}
 		else {
-			if (options.track_opcode_stats) {
+			_impl->packet_sent_count[p->GetOpcode()]++;
-				//m_packet_sent_count[p->GetOpcode()]++; //Wont bother with bypass tracking of these since those are rare for testing anyway
-			}
 			opcode = (*_impl->opcode_manager)->EmuToEQ(p->GetOpcode());
 		}
 
@@ -394,7 +387,7 @@ void EQ::Net::ConcurrentEQStream::QueuePacket(const EQApplicationPacket *p, bool
 			break;
 		}
 
-		ConcurrentEQStreamQueuePacketMessage msg;
+		ceqs_queue_packet_msg_t msg;
 		msg.type = ceqs_msg_type::QueuePacket;
 		msg.stream_id = _impl->id;
 		msg.packet = out;
@@ -402,16 +395,15 @@ void EQ::Net::ConcurrentEQStream::QueuePacket(const EQApplicationPacket *p, bool
 
 		//Make sure the background gets this message
 		_impl->parent->_PushToBackgroundQueue((ceqs_msg_t*)&msg);
-
-		printf("(foreground) Packet Queue for %u with %u bytes with ack: %s\n", _impl->id, out->Length(), ack_req ? "true" : "false");
 	}
 }
 
 //Called by foreground
 void EQ::Net::ConcurrentEQStream::FastQueuePacket(EQApplicationPacket **p, bool ack_req)
 {
-	std::unique_ptr<EQApplicationPacket> app(*p);
+	QueuePacket(*p, ack_req);
-	QueuePacket(app.get(), ack_req);
+	delete *p;
+	*p = nullptr;
 }
 
 //Called by foreground
@@ -440,9 +432,7 @@ EQApplicationPacket *EQ::Net::ConcurrentEQStream::PopPacket()
 		}
 
 		EmuOpcode emu_op = (*_impl->opcode_manager)->EQToEmu(opcode);
-		if (options.track_opcode_stats) {
+		_impl->packet_recv_count[emu_op]++;
-			//m_packet_recv_count[emu_op]++;
-		}
 
 		EQApplicationPacket *ret = new EQApplicationPacket(emu_op, (unsigned char*)p->Data() + options.opcode_size, p->Length() - options.opcode_size);
 		ret->SetProtocolOpcode(opcode);
@@ -460,7 +450,7 @@ void EQ::Net::ConcurrentEQStream::Close()
 		return;
 	}
 
-	ConcurrentEQStreamCloseConnectionMessage msg;
+	ceqs_close_connection_msg_t msg;
 	msg.type = CloseConnection;
 	msg.stream_id = _impl->id;
 
@@ -554,24 +544,24 @@ EQStreamInterface::MatchState EQ::Net::ConcurrentEQStream::CheckSignature(const
 
 		if (opcode == sig->first_eq_opcode) {
 			if (length == sig->first_length) {
-			//	LogF(Logs::General, Logs::Netcode, "[IDENT_TRACE] {0}:{1}: First opcode matched {2:#x} and length matched {3}",
+				LogF(Logs::General, Logs::Netcode, "[IDENT_TRACE] {0}:{1}: First opcode matched {2:#x} and length matched {3}",
-			//		m_connection->RemoteEndpoint(), m_connection->RemotePort(), sig->first_eq_opcode, length);
+					GetRemoteAddr(), GetRemotePort(), sig->first_eq_opcode, length);
 				return MatchSuccessful;
 			}
 			else if (length == 0) {
-			//	LogF(Logs::General, Logs::Netcode, "[IDENT_TRACE] {0}:{1}: First opcode matched {2:#x} and length is ignored.",
+				LogF(Logs::General, Logs::Netcode, "[IDENT_TRACE] {0}:{1}: First opcode matched {2:#x} and length is ignored.",
-			//		m_connection->RemoteEndpoint(), m_connection->RemotePort(), sig->first_eq_opcode);
+					GetRemoteAddr(), GetRemotePort(), sig->first_eq_opcode);
 				return MatchSuccessful;
 			}
 			else {
-			//	LogF(Logs::General, Logs::Netcode, "[IDENT_TRACE] {0}:{1}: First opcode matched {2:#x} but length {3} did not match expected {4}",
+				LogF(Logs::General, Logs::Netcode, "[IDENT_TRACE] {0}:{1}: First opcode matched {2:#x} but length {3} did not match expected {4}",
-			//		m_connection->RemoteEndpoint(), m_connection->RemotePort(), sig->first_eq_opcode, length, sig->first_length);
+					GetRemoteAddr(), GetRemotePort(), sig->first_eq_opcode, length, sig->first_length);
 				return MatchFailed;
 			}
 		}
 		else {
-			//LogF(Logs::General, Logs::Netcode, "[IDENT_TRACE] {0}:{1}: First opcode {1:#x} did not match expected {2:#x}",
+			LogF(Logs::General, Logs::Netcode, "[IDENT_TRACE] {0}:{1}: First opcode {1:#x} did not match expected {2:#x}",
-			//	m_connection->RemoteEndpoint(), m_connection->RemotePort(), opcode, sig->first_eq_opcode);
+				GetRemoteAddr(), GetRemotePort(), opcode, sig->first_eq_opcode);
 			return MatchFailed;
 		}
 	}
@@ -605,6 +595,20 @@ EQStreamInterface::Stats EQ::Net::ConcurrentEQStream::GetStats() const
 {
 	EQStreamInterface::Stats ret;
 	ret.DaybreakStats = _impl->stats;
+
+	for (int i = 0; i < _maxEmuOpcode; ++i) {
+		ret.RecvCount[i] = 0;
+		ret.SentCount[i] = 0;
+	}
+
+	for (auto &s : _impl->packet_sent_count) {
+		ret.SentCount[s.first] = s.second;
+	}
+
+	for (auto &r : _impl->packet_recv_count) {
+		ret.RecvCount[r.first] = r.second;
+	}
+
 	return ret;
 }
 
@@ -615,7 +619,7 @@ void EQ::Net::ConcurrentEQStream::ResetStats()
 		return;
 	}
 
-	ConcurrentEQStreamResetStatsMessage msg;
+	ceqs_reset_stats_msg_t msg;
 	msg.type = ceqs_msg_type::ResetStats;
 	msg.stream_id = _impl->id;
 

common/net/eqstream_concurrent_message.h

@@ -0,0 +1,89 @@
+#pragma once
+
+#define EQSM_PAD_LEN 252
+
+namespace EQ
+{
+	namespace Net
+	{
+		class DynamicPacket;
+		enum ceqs_msg_type : uint32_t
+		{
+			//Sent by background
+			NewConnection,
+			ConnectionStateChange,
+			PacketRecv,
+			UpdateStats,
+			//Sent by foreground
+			QueuePacket,
+			TerminateBackground,
+			CloseConnection,
+			ResetStats
+		};
+
+		typedef struct  
+		{
+			ceqs_msg_type type;
+			char padding[EQSM_PAD_LEN];
+		} ceqs_msg_t;
+
+		//Sent by background
+		typedef struct
+		{
+			ceqs_msg_type type;
+			uint64_t stream_id;
+			int remote_port;
+			int state;
+			char endpoint[64];
+		} ceqs_new_connection_msg_t;
+
+		typedef struct
+		{
+			ceqs_msg_type type;
+			uint64_t stream_id;
+			int from;
+			int to;
+		} ceqs_connection_state_change_msg_t;
+
+		typedef struct
+		{
+			ceqs_msg_type type;
+			uint64_t stream_id;
+			EQ::Net::DynamicPacket *packet;
+		} ceqs_packet_recv_msg_t;
+
+		typedef struct
+		{
+			ceqs_msg_type type;
+			uint64_t stream_id;
+			DaybreakConnectionStats stats;
+		} ceqs_update_stats_msg_t;
+
+		//Sent by foreground
+		typedef struct
+		{
+			ceqs_msg_type type;
+			uint64_t stream_id;
+			EQ::Net::DynamicPacket *packet;
+			bool ack_req;
+		} ceqs_queue_packet_msg_t;
+
+		typedef struct
+		{
+			ceqs_msg_type type;
+			uint64_t stream_id;
+		} ceqs_close_connection_msg_t;
+
+		typedef struct
+		{
+			ceqs_msg_type type;
+			uint64_t stream_id;
+		} ceqs_reset_stats_msg_t;
+
+		typedef struct
+		{
+			ceqs_msg_type type;
+		} ceqs_terminate_msg_t;
+
+	}
+}

loginserver/client_manager.cpp

@@ -28,7 +28,7 @@ ClientManager::ClientManager()
 {
 	int titanium_port = atoi(server.config->GetVariable("Titanium", "port").c_str());
 	EQStreamManagerInterfaceOptions titanium_opts(titanium_port, false, false);
-	titanium_stream = new EQ::Net::EQStreamManager(titanium_opts);
+	titanium_stream = new EQ::Net::ConcurrentEQStreamManager(titanium_opts);
 	titanium_ops = new RegularOpcodeManager;
 	if (!titanium_ops->LoadOpcodes(server.config->GetVariable("Titanium", "opcodes").c_str()))
 	{
@@ -38,7 +38,7 @@ ClientManager::ClientManager()
 	}
 
 	titanium_stream->OnNewConnection([this](std::shared_ptr<EQStreamInterface> stream) {
-		LogF(Logs::General, Logs::Login_Server, "New Titanium client connection from {0}:{1}", stream->GetRemoteIP(), stream->GetRemotePort());
+		LogF(Logs::General, Logs::Login_Server, "New Titanium client connection from {0}:{1}", stream->GetRemoteAddr(), stream->GetRemotePort());
 		stream->SetOpcodeManager(&titanium_ops);
 		Client *c = new Client(stream, cv_titanium);
 		clients.push_back(c);
@@ -46,7 +46,7 @@ ClientManager::ClientManager()
 
 	int sod_port = atoi(server.config->GetVariable("SoD", "port").c_str());
 	EQStreamManagerInterfaceOptions sod_opts(sod_port, false, false);
-	sod_stream = new EQ::Net::EQStreamManager(sod_opts);
+	sod_stream = new EQ::Net::ConcurrentEQStreamManager(sod_opts);
 	sod_ops = new RegularOpcodeManager;
 	if (!sod_ops->LoadOpcodes(server.config->GetVariable("SoD", "opcodes").c_str()))
 	{
@@ -56,7 +56,7 @@ ClientManager::ClientManager()
 	}
 
 	sod_stream->OnNewConnection([this](std::shared_ptr<EQStreamInterface> stream) {
-		LogF(Logs::General, Logs::Login_Server, "New SoD client connection from {0}:{1}", stream->GetRemoteIP(), stream->GetRemotePort());
+		LogF(Logs::General, Logs::Login_Server, "New SoD client connection from {0}:{1}", stream->GetRemoteAddr(), stream->GetRemotePort());
 		stream->SetOpcodeManager(&sod_ops);
 		Client *c = new Client(stream, cv_sod);
 		clients.push_back(c);

loginserver/client_manager.h

@@ -20,7 +20,7 @@
 
 #include "../common/global_define.h"
 #include "../common/opcodemgr.h"
-#include "../common/net/eqstream.h"
+#include "../common/net/eqstream_concurrent.h"
 #include "client.h"
 #include <list>
 
@@ -63,9 +63,9 @@ private:
 
 	std::list<Client*> clients;
 	OpcodeManager *titanium_ops;
-	EQ::Net::EQStreamManager *titanium_stream;
+	EQ::Net::ConcurrentEQStreamManager *titanium_stream;
 	OpcodeManager *sod_ops;
-	EQ::Net::EQStreamManager *sod_stream;
+	EQ::Net::ConcurrentEQStreamManager *sod_stream;
 };
 
 #endif

zone/command.cpp

@@ -9528,26 +9528,23 @@ void command_netstats(Client *c, const Seperator *sep)
 			c->Message(0, "Outgoing Link Saturation %.2f%% (%.2fkb/sec)", 100.0 * (1.0 - ((opts.daybreak_options.outgoing_data_rate - stats.datarate_remaining) / opts.daybreak_options.outgoing_data_rate)), opts.daybreak_options.outgoing_data_rate);
 		}
 		
-		if (opts.track_opcode_stats) {
+		c->Message(0, "--------------------------------------------------------------------");
-			c->Message(0, "--------------------------------------------------------------------");
+		c->Message(0, "Sent Packet Types");
-			c->Message(0, "Sent Packet Types");
+		for (auto i = 0; i < _maxEmuOpcode; ++i) {
-			for (auto i = 0; i < _maxEmuOpcode; ++i) {
+			auto cnt = eqs_stats.SentCount[i];
-				auto cnt = eqs_stats.SentCount[i];
+			if (cnt > 0) {
-				if (cnt > 0) {
+				c->Message(0, "%s: %u (%.2f / sec)", OpcodeNames[i], cnt, cnt / sec_since_stats_reset);
-					c->Message(0, "%s: %u (%.2f / sec)", OpcodeNames[i], cnt, cnt / sec_since_stats_reset);
+			}
-				}
+		}
-			}
+
-
+		c->Message(0, "--------------------------------------------------------------------");
-			c->Message(0, "--------------------------------------------------------------------");
+		c->Message(0, "Recv Packet Types");
-			c->Message(0, "Recv Packet Types");
+		for (auto i = 0; i < _maxEmuOpcode; ++i) {
-			for (auto i = 0; i < _maxEmuOpcode; ++i) {
+			auto cnt = eqs_stats.RecvCount[i];
-				auto cnt = eqs_stats.RecvCount[i];
+			if (cnt > 0) {
-				if (cnt > 0) {
+				c->Message(0, "%s: %u (%.2f / sec)", OpcodeNames[i], cnt, cnt / sec_since_stats_reset);
-					c->Message(0, "%s: %u (%.2f / sec)", OpcodeNames[i], cnt, cnt / sec_since_stats_reset);
-				}
 			}
 		}
-		
 		c->Message(0, "--------------------------------------------------------------------");
 	}
 }

zone/eqemu_api_zone_data_service.cpp

@@ -691,30 +691,27 @@ void callGetPacketStatistics(Json::Value &response)
 		row["resent_non_fragments"]     = stats.resent_full;
 		row["dropped_datarate_packets"] = stats.dropped_datarate_packets;
 
-		if (opts.track_opcode_stats) {
+		Json::Value sent_packet_types;
 
-			Json::Value sent_packet_types;
+		for (auto i = 0; i < _maxEmuOpcode; ++i) {
-
+			auto count = eqs_stats.SentCount[i];
-			for (auto i = 0; i < _maxEmuOpcode; ++i) {
+			if (count > 0) {
-				auto count = eqs_stats.SentCount[i];
+				sent_packet_types[OpcodeNames[i]] = count;
-				if (count > 0) {
-					sent_packet_types[OpcodeNames[i]] = count;
-				}
 			}
-
-			Json::Value receive_packet_types;
-
-			for (auto i = 0; i < _maxEmuOpcode; ++i) {
-				auto count = eqs_stats.RecvCount[i];
-				if (count > 0) {
-					receive_packet_types[OpcodeNames[i]] = count;
-				}
-			}
-
-			row["sent_packet_types"]    = sent_packet_types;
-			row["receive_packet_types"] = receive_packet_types;
 		}
 
+		Json::Value receive_packet_types;
+
+		for (auto i = 0; i < _maxEmuOpcode; ++i) {
+			auto count = eqs_stats.RecvCount[i];
+			if (count > 0) {
+				receive_packet_types[OpcodeNames[i]] = count;
+			}
+		}
+
+		row["sent_packet_types"]    = sent_packet_types;
+		row["receive_packet_types"] = receive_packet_types;
+
 		response.append(row);
 	}
 }
@@ -776,4 +773,4 @@ void EQEmuApiZoneDataService::get(Json::Value &response, const std::vector<std::
 	if (method == "get_zone_attributes") {
 		callGetZoneAttributes(response);
 	}
-}
+}

zone/net.cpp

@@ -44,6 +44,7 @@ Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
 #include "../common/eqemu_logsys.h"
 #include "../common/eqemu_logsys_fmt.h"
 #include "../common/net/console_server.h"
+#include "../common/net/eqstream_concurrent.h"
 
 #include "zone_config.h"
 #include "masterentity.h"
@@ -461,7 +462,7 @@ int main(int argc, char** argv) {
 	UpdateWindowTitle();
 	std::shared_ptr<EQStreamInterface> eqss;
 	EQStreamInterface *eqsi;
-	std::unique_ptr<EQ::Net::EQStreamManager> eqsm;
+	std::unique_ptr<EQ::Net::ConcurrentEQStreamManager> eqsm;
 	std::chrono::time_point<std::chrono::system_clock> frame_prev = std::chrono::system_clock::now();
 	std::unique_ptr<EQ::Net::ConsoleServer> console;
 
@@ -506,8 +507,7 @@ int main(int argc, char** argv) {
 			opts.daybreak_options.resend_delay_min = RuleI(Network, ResendDelayMinMS);
 			opts.daybreak_options.resend_delay_max = RuleI(Network, ResendDelayMaxMS);
 			opts.daybreak_options.outgoing_data_rate = RuleR(Network, ClientDataRate);
-			opts.track_opcode_stats = RuleB(Network, TrackOpcodeStats);
+			eqsm.reset(new EQ::Net::ConcurrentEQStreamManager(opts));
-			eqsm.reset(new EQ::Net::EQStreamManager(opts));
 			eqsf_open = true;
 
 			eqsm->OnNewConnection([&stream_identifier](std::shared_ptr<EQStreamInterface> stream) {