From a76149c8e300f28ad186534ceb9e7dc0518f9432 Mon Sep 17 00:00:00 2001 From: KimLS Date: Wed, 28 Sep 2016 23:51:37 -0700 Subject: [PATCH] Some work on compression, the way the client does it is... bizarre and not how i orig thought it would be. --- common/net/daybreak_connection.cpp | 102 +++++++++----- common/net/daybreak_connection.h | 13 +- common/net/daybreak_structs.h | 8 +- common/net/eqstream.h | 21 +-- world/net.cpp | 8 +- zone/CMakeLists.txt | 16 +-- zone/net.cpp | 216 ++++++++++++++--------------- 7 files changed, 200 insertions(+), 184 deletions(-) diff --git a/common/net/daybreak_connection.cpp b/common/net/daybreak_connection.cpp index 709b4611c..5f392b664 100644 --- a/common/net/daybreak_connection.cpp +++ b/common/net/daybreak_connection.cpp @@ -41,12 +41,12 @@ void EQ::Net::DaybreakConnectionManager::Attach(uv_loop_t *loop) uv_timer_start(&m_timer, [](uv_timer_t *handle) { DaybreakConnectionManager *c = (DaybreakConnectionManager*)handle->data; c->Process(); - }, 5, 5); + }, 2, 2); uv_timer_start(&m_resend_timer, [](uv_timer_t *handle) { DaybreakConnectionManager *c = (DaybreakConnectionManager*)handle->data; c->ProcessResend(); - }, 50, 50); + }, 10, 10); uv_udp_init(loop, &m_socket); m_socket.data = this; @@ -88,6 +88,16 @@ void EQ::Net::DaybreakConnectionManager::Detach() void EQ::Net::DaybreakConnectionManager::Connect(const std::string &addr, int port) { + //todo dns resolution + + auto connection = std::shared_ptr(new DaybreakConnection(this, addr, port)); + connection->m_self = connection; + + if (m_on_new_connection) { + m_on_new_connection(connection); + } + + m_connections.insert(std::make_pair(std::make_pair(addr, port), connection)); } void EQ::Net::DaybreakConnectionManager::Process() @@ -252,6 +262,7 @@ EQ::Net::DaybreakConnection::DaybreakConnection(DaybreakConnectionManager *owner m_buffered_packets_length = 0; m_resend_delay = m_owner->m_options.resend_delay_ms; m_rolling_ping = 100; + m_last_session_stats = Clock::now(); } //new connection made as client @@ -271,6 +282,7 @@ EQ::Net::DaybreakConnection::DaybreakConnection(DaybreakConnectionManager *owner m_buffered_packets_length = 0; m_resend_delay = m_owner->m_options.resend_delay_ms; m_rolling_ping = 100; + m_last_session_stats = Clock::now(); } EQ::Net::DaybreakConnection::~DaybreakConnection() @@ -284,7 +296,6 @@ void EQ::Net::DaybreakConnection::Close() disconnect.zero = 0; disconnect.opcode = OP_SessionDisconnect; disconnect.connect_code = HostToNetwork(m_connect_code); - disconnect.disconnect_code = 6; WritablePacket out; out.PutSerialize(0, disconnect); InternalSend(out); @@ -344,7 +355,7 @@ void EQ::Net::DaybreakConnection::Process() ProcessQueue(); auto time_since_stats = (size_t)std::chrono::duration_cast(now - m_last_stats).count(); - if (time_since_stats >= m_owner->m_options.stats_delay_ms) { + if (m_owner->m_options.stats_delay_ms > 0 && time_since_stats >= m_owner->m_options.stats_delay_ms) { SendStatSync(); m_last_stats = now; } @@ -376,40 +387,24 @@ void EQ::Net::DaybreakConnection::ProcessPacket(Packet &p) return; } - if (m_encode_passes[0] == EncodeCompression || m_encode_passes[1] == EncodeCompression) { - EQ::Net::WritablePacket temp; - temp.PutPacket(0, p); + EQ::Net::WritablePacket temp; + temp.PutPacket(0, p); + temp.Resize(temp.Length() - m_crc_bytes); - for (int i = 0; i < 2; ++i) { - switch (m_encode_passes[i]) { - case EncodeCompression: - Log.OutF(Logs::General, Logs::Debug, "Decompressing packet on pass {1}\n{0}", temp.ToString(), i); - Decompress(temp, DaybreakHeader::size(), temp.Length() - DaybreakHeader::size() - m_crc_bytes); - Log.OutF(Logs::General, Logs::Debug, "Decompressed packet on pass {1}\n{0}", temp.ToString(), i); - break; - case EncodeXOR: - Decode(temp, DaybreakHeader::size(), temp.Length() - DaybreakHeader::size() - m_crc_bytes); - break; - default: - break; - } + for (int i = 1; i >= 0; --i) { + switch (m_encode_passes[i]) { + case EncodeCompression: + Decompress(temp, DaybreakHeader::size(), temp.Length() - DaybreakHeader::size()); + break; + case EncodeXOR: + Decode(temp, DaybreakHeader::size(), temp.Length() - DaybreakHeader::size()); + break; + default: + break; } - - ProcessDecodedPacket(ReadOnlyPacket(temp.Data(), temp.Length() - m_crc_bytes)); } - else { - for (int i = 0; i < 2; ++i) { - switch (m_encode_passes[i]) { - case EncodeXOR: - Decode(p, DaybreakHeader::size(), p.Length() - DaybreakHeader::size() - m_crc_bytes); - break; - default: - break; - } - } - ProcessDecodedPacket(ReadOnlyPacket(p.Data(), p.Length() - m_crc_bytes)); - } + ProcessDecodedPacket(ReadOnlyPacket(temp.Data(), temp.Length())); } else { ProcessDecodedPacket(p); @@ -667,7 +662,6 @@ void EQ::Net::DaybreakConnection::ProcessDecodedPacket(const Packet &p) disconnect.zero = 0; disconnect.opcode = OP_SessionDisconnect; disconnect.connect_code = HostToNetwork(m_connect_code); - disconnect.disconnect_code = 6; WritablePacket out; out.PutSerialize(0, disconnect); InternalSend(out); @@ -685,6 +679,24 @@ void EQ::Net::DaybreakConnection::ProcessDecodedPacket(const Packet &p) } break; } + case OP_SessionStatRequest: + { + auto request = p.GetSerialize(0); + + DaybreakSessionStatResponse response; + response.zero = 0; + response.opcode = OP_SessionStatResponse; + response.timestamp = request.timestamp; + response.our_timestamp = EQ::Net::HostToNetwork(std::chrono::duration_cast(std::chrono::system_clock::now().time_since_epoch()).count()); + response.client_sent = request.packets_sent; + response.client_recv = request.packets_recv; + response.server_sent = EQ::Net::HostToNetwork(m_stats.sent_packets); + response.server_recv = EQ::Net::HostToNetwork(m_stats.recv_packets); + WritablePacket out; + out.PutSerialize(0, response); + InternalSend(out); + break; + } default: Log.OutF(Logs::Detail, Logs::Netcode, "Unhandled opcode {0:#x}", p.GetInt8(1)); break; @@ -916,11 +928,13 @@ void EQ::Net::DaybreakConnection::Compress(Packet &p, size_t offset, size_t leng uint32_t new_length = 0; if (length > 30) { - new_length = Deflate(buffer, (uint32_t)length, new_buffer + 1, 2048); + new_length = Deflate(buffer, (uint32_t)length, new_buffer + 1, 2048) + 1; new_buffer[0] = 0x5a; } else { + memcpy(new_buffer + 1, buffer, length); new_buffer[0] = 0xa5; + new_length = length + 1; } p.Resize(offset); @@ -1013,7 +1027,21 @@ void EQ::Net::DaybreakConnection::SendOutOfOrderAck(int stream_id, uint16_t seq) void EQ::Net::DaybreakConnection::SendStatSync() { - //todo + DaybreakSessionStatRequest request; + request.zero = 0; + request.opcode = OP_SessionStatRequest; + request.timestamp = EQ::Net::HostToNetwork(std::chrono::duration_cast(std::chrono::system_clock::now().time_since_epoch()).count() & 0xFFFFLL); + request.stat_ping = m_stats.last_stat_ping; + if (m_stats.total_stat_count > 0) + request.avg_ping = m_stats.total_stat_ping / m_stats.total_stat_count; + else + request.avg_ping = 0; + request.min_ping = m_stats.min_stat_ping; + request.max_ping = m_stats.max_stat_ping; + request.last_ping = m_stats.last_stat_ping; + request.packets_sent = m_stats.sent_packets + 1; + request.packets_recv = m_stats.recv_packets; + m_last_session_stats = Clock::now(); } void EQ::Net::DaybreakConnection::InternalBufferedSend(Packet &p) diff --git a/common/net/daybreak_connection.h b/common/net/daybreak_connection.h index 76f6527e5..762536ce1 100644 --- a/common/net/daybreak_connection.h +++ b/common/net/daybreak_connection.h @@ -82,6 +82,11 @@ namespace EQ total_acks = 0; min_ping = 0xFFFFFFFFFFFFFFFFUL; max_ping = 0; + total_stat_ping = 0; + total_stat_count = 0; + min_stat_ping = 0xFFFFFFFFFFFFFFFFUL; + max_stat_ping = 0; + last_stat_ping = 0; created = Clock::now(); } @@ -93,6 +98,11 @@ namespace EQ uint64_t total_acks; uint64_t min_ping; uint64_t max_ping; + uint64_t total_stat_ping; + uint64_t total_stat_count; + uint64_t min_stat_ping; + uint64_t max_stat_ping; + uint64_t last_stat_ping; Timestamp created; }; @@ -135,7 +145,6 @@ namespace EQ Timestamp m_last_stats; DaybreakConnectionStats m_stats; Timestamp m_last_session_stats; - uint64_t m_last_session_stats_ping; size_t m_resend_delay; size_t m_rolling_ping; @@ -209,7 +218,7 @@ namespace EQ max_connection_count = 0; keepalive_delay_ms = 0; resend_delay_ms = 1000; - stats_delay_ms = 10000; + stats_delay_ms = 0; connect_delay_ms = 1000; stale_connection_ms = 60000; crc_length = 2; diff --git a/common/net/daybreak_structs.h b/common/net/daybreak_structs.h index eb3cba744..36e97a246 100644 --- a/common/net/daybreak_structs.h +++ b/common/net/daybreak_structs.h @@ -74,15 +74,13 @@ namespace EQ uint8_t zero; uint8_t opcode; uint32_t connect_code; - uint16_t disconnect_code; template void serialize(Archive & archive) { archive(CEREAL_NVP(zero), CEREAL_NVP(opcode), - CEREAL_NVP(connect_code), - CEREAL_NVP(disconnect_code)); + CEREAL_NVP(connect_code)); } }; @@ -116,7 +114,7 @@ namespace EQ } }; - struct DaybreakSessionStatRequestHeader + struct DaybreakSessionStatRequest { static size_t size() { return 40; } uint8_t zero; @@ -146,7 +144,7 @@ namespace EQ } }; - struct DaybreakSessionStatResponseHeader + struct DaybreakSessionStatResponse { static size_t size() { return 40; } uint8_t zero; diff --git a/common/net/eqstream.h b/common/net/eqstream.h index 04f523671..a24aeacec 100644 --- a/common/net/eqstream.h +++ b/common/net/eqstream.h @@ -17,20 +17,21 @@ namespace EQ opcode_size = 2; } - EQStreamManagerOptions(bool encoded, bool compressed) { + EQStreamManagerOptions(int port, bool encoded, bool compressed) { opcode_size = 2; - if (encoded) { - daybreak_options.encode_passes[0] = EncodeXOR; - if (compressed) { - daybreak_options.encode_passes[1] = EncodeCompression; - } + //World seems to support both compression and xor zone supports one or the others. + //Enforce one or the other in the convienence construct + //Login I had trouble getting to recognize compression at all + //but that might be because it was still a bit buggy when i was testing that. + if (compressed) { + daybreak_options.encode_passes[0] = EncodeCompression; } - else { - if (compressed) { - daybreak_options.encode_passes[0] = EncodeCompression; - } + else if (encoded) { + daybreak_options.encode_passes[0] = EncodeXOR; } + + daybreak_options.port = port; } int opcode_size; diff --git a/world/net.cpp b/world/net.cpp index 02492ef30..245d09274 100644 --- a/world/net.cpp +++ b/world/net.cpp @@ -116,9 +116,6 @@ void CatchSignal(int sig_num); int main(int argc, char** argv) { RegisterExecutablePlatform(ExePlatformWorld); Log.LoadLogSettingsDefaults(); - - - set_exception_handler(); /* Database Version Check */ @@ -398,10 +395,7 @@ int main(int argc, char** argv) { return 1; } - EQ::Net::EQStreamManagerOptions opts(false, true); - opts.daybreak_options.port = 9000; - opts.opcode_size = 2; - + EQ::Net::EQStreamManagerOptions opts(9000, false, true); EQ::Net::EQStreamManager eqsm(opts); //register all the patches we have avaliable with the stream identifier. diff --git a/zone/CMakeLists.txt b/zone/CMakeLists.txt index d424bd03c..019f9ab7f 100644 --- a/zone/CMakeLists.txt +++ b/zone/CMakeLists.txt @@ -231,7 +231,11 @@ INSTALL(TARGETS zone RUNTIME DESTINATION ${CMAKE_INSTALL_PREFIX}/bin) ADD_DEFINITIONS(-DZONE) -TARGET_LINK_LIBRARIES(zone common debug ${MySQL_LIBRARY_DEBUG} optimized ${MySQL_LIBRARY_RELEASE} ${ZLIB_LIBRARY}) +TARGET_LINK_LIBRARIES(zone common debug ${MySQL_LIBRARY_DEBUG} optimized ${MySQL_LIBRARY_RELEASE} ${ZLIB_LIBRARY} libuv fmt) + +IF(WIN32) + TARGET_LINK_LIBRARIES(zone "ws2_32" "psapi" "iphlpapi" "userenv") +ENDIF(WIN32) IF(EQEMU_BUILD_PERL) TARGET_LINK_LIBRARIES(zone ${PERL_LIBRARY}) @@ -241,16 +245,6 @@ IF(EQEMU_BUILD_LUA) TARGET_LINK_LIBRARIES(zone luabind ${LUA_LIBRARY}) ENDIF(EQEMU_BUILD_LUA) - -IF(MSVC) - SET_TARGET_PROPERTIES(zone PROPERTIES LINK_FLAGS_RELEASE "/OPT:REF /OPT:ICF") - TARGET_LINK_LIBRARIES(zone "Ws2_32.lib") -ENDIF(MSVC) - -IF(MINGW) - TARGET_LINK_LIBRARIES(zone "WS2_32") -ENDIF(MINGW) - IF(UNIX) TARGET_LINK_LIBRARIES(zone "${CMAKE_DL_LIBS}") TARGET_LINK_LIBRARIES(zone "z") diff --git a/zone/net.cpp b/zone/net.cpp index 20a37114d..a02713457 100644 --- a/zone/net.cpp +++ b/zone/net.cpp @@ -65,6 +65,10 @@ #include "lua_parser.h" #include "questmgr.h" +#include "../common/event/event_loop.h" +#include "../common/event/timer.h" +#include "../common/net/eqstream.h" + #include #include #include @@ -73,6 +77,7 @@ #include #include #include +#include #ifdef _CRTDBG_MAP_ALLOC #undef new @@ -97,7 +102,6 @@ WorldServer worldserver; uint32 numclients = 0; char errorname[32]; extern Zone* zone; -EQStreamFactory eqsf(ZoneStream); npcDecayTimes_Struct npcCorpseDecayTimes[100]; TitleManager title_manager; QueryServ *QServ = 0; @@ -431,122 +435,111 @@ int main(int argc, char** argv) { uint8 ZONEUPDATE = 10; Timer zoneupdate_timer(ZONEUPDATE); zoneupdate_timer.Start(); - while(RunLoops) { - { //profiler block to omit the sleep from times + bool eqsf_open = false; + std::unique_ptr eqsm; - //Advance the timer to our current point in time - Timer::SetCurrentTime(); - - worldserver.Process(); - - if (!eqsf.IsOpen() && Config->ZonePort != 0) { - Log.Out(Logs::General, Logs::Zone_Server, "Starting EQ Network server on port %d", Config->ZonePort); - if (!eqsf.Open(Config->ZonePort)) { - Log.Out(Logs::General, Logs::Error, "Failed to open port %d", Config->ZonePort); - ZoneConfig::SetZonePort(0); - worldserver.Disconnect(); + EQ::Timer process_timer(33, true, [&eqsf_open, &eqsm, &stream_identifier, &eqsi, &worldwasconnected, + &zoneupdate_timer, &IDLEZONEUPDATE, &ZONEUPDATE, &quest_timers, &InterserverTimer]() { + //Advance the timer to our current point in time + Timer::SetCurrentTime(); + + worldserver.Process(); + + if (!eqsf_open && Config->ZonePort != 0) { + Log.Out(Logs::General, Logs::Zone_Server, "Starting EQ Network server on port %d", Config->ZonePort); + + EQ::Net::EQStreamManagerOptions opts(Config->ZonePort, false, true); + eqsm.reset(new EQ::Net::EQStreamManager(opts)); + eqsf_open = true; + + eqsm->OnNewConnection([&stream_identifier](std::shared_ptr stream) { + stream_identifier.AddStream(stream); + Log.OutF(Logs::Detail, Logs::World_Server, "New connection from IP {0}:{1}", stream->RemoteEndpoint(), ntohs(stream->GetRemotePort())); + }); + } + + //give the stream identifier a chance to do its work.... + stream_identifier.Process(); + + //check the stream identifier for any now-identified streams + while((eqsi = stream_identifier.PopIdentified())) { + //now that we know what patch they are running, start up their client object + struct in_addr in; + in.s_addr = eqsi->GetRemoteIP(); + Log.Out(Logs::Detail, Logs::World_Server, "New client from %s:%d", inet_ntoa(in), ntohs(eqsi->GetRemotePort())); + auto client = new Client(eqsi); + entity_list.AddClient(client); + } + + if ( numclients < 1 && zoneupdate_timer.GetDuration() != IDLEZONEUPDATE ) + zoneupdate_timer.SetTimer(IDLEZONEUPDATE); + else if ( numclients > 0 && zoneupdate_timer.GetDuration() == IDLEZONEUPDATE ) + { + zoneupdate_timer.SetTimer(ZONEUPDATE); + zoneupdate_timer.Trigger(); + } + + //check for timeouts in other threads + timeout_manager.CheckTimeouts(); + + if (worldserver.Connected()) { + worldwasconnected = true; + } + else { + if (worldwasconnected && is_zone_loaded) + entity_list.ChannelMessageFromWorld(0, 0, 6, 0, 0, "WARNING: World server connection lost"); worldwasconnected = false; } - } - - //check the factory for any new incoming streams. - while ((eqss = eqsf.Pop())) { - //pull the stream out of the factory and give it to the stream identifier - //which will figure out what patch they are running, and set up the dynamic - //structures and opcodes for that patch. - struct in_addr in; - in.s_addr = eqss->GetRemoteIP(); - Log.Out(Logs::Detail, Logs::World_Server, "New connection from %s:%d", inet_ntoa(in), ntohs(eqss->GetRemotePort())); - stream_identifier.AddStream(eqss); //takes the stream - } - - //give the stream identifier a chance to do its work.... - stream_identifier.Process(); - - //check the stream identifier for any now-identified streams - while((eqsi = stream_identifier.PopIdentified())) { - //now that we know what patch they are running, start up their client object - struct in_addr in; - in.s_addr = eqsi->GetRemoteIP(); - Log.Out(Logs::Detail, Logs::World_Server, "New client from %s:%d", inet_ntoa(in), ntohs(eqsi->GetRemotePort())); - auto client = new Client(eqsi); - entity_list.AddClient(client); - } - - if ( numclients < 1 && zoneupdate_timer.GetDuration() != IDLEZONEUPDATE ) - zoneupdate_timer.SetTimer(IDLEZONEUPDATE); - else if ( numclients > 0 && zoneupdate_timer.GetDuration() == IDLEZONEUPDATE ) - { - zoneupdate_timer.SetTimer(ZONEUPDATE); - zoneupdate_timer.Trigger(); - } - - //check for timeouts in other threads - timeout_manager.CheckTimeouts(); - - if (worldserver.Connected()) { - worldwasconnected = true; - } - else { - if (worldwasconnected && is_zone_loaded) - entity_list.ChannelMessageFromWorld(0, 0, 6, 0, 0, "WARNING: World server connection lost"); - worldwasconnected = false; - } - - if (is_zone_loaded && zoneupdate_timer.Check()) { - { - if(net.group_timer.Enabled() && net.group_timer.Check()) - entity_list.GroupProcess(); - - if(net.door_timer.Enabled() && net.door_timer.Check()) - entity_list.DoorProcess(); - - if(net.object_timer.Enabled() && net.object_timer.Check()) - entity_list.ObjectProcess(); - - if(net.corpse_timer.Enabled() && net.corpse_timer.Check()) - entity_list.CorpseProcess(); - - if(net.trap_timer.Enabled() && net.trap_timer.Check()) - entity_list.TrapProcess(); - - if(net.raid_timer.Enabled() && net.raid_timer.Check()) - entity_list.RaidProcess(); - - entity_list.Process(); - entity_list.MobProcess(); - entity_list.BeaconProcess(); - entity_list.EncounterProcess(); - - if (zone) { - if(!zone->Process()) { - Zone::Shutdown(); + + if (is_zone_loaded && zoneupdate_timer.Check()) { + { + if(net.group_timer.Enabled() && net.group_timer.Check()) + entity_list.GroupProcess(); + + if(net.door_timer.Enabled() && net.door_timer.Check()) + entity_list.DoorProcess(); + + if(net.object_timer.Enabled() && net.object_timer.Check()) + entity_list.ObjectProcess(); + + if(net.corpse_timer.Enabled() && net.corpse_timer.Check()) + entity_list.CorpseProcess(); + + if(net.trap_timer.Enabled() && net.trap_timer.Check()) + entity_list.TrapProcess(); + + if(net.raid_timer.Enabled() && net.raid_timer.Check()) + entity_list.RaidProcess(); + + entity_list.Process(); + entity_list.MobProcess(); + entity_list.BeaconProcess(); + entity_list.EncounterProcess(); + + if (zone) { + if(!zone->Process()) { + Zone::Shutdown(); + } } + + if(quest_timers.Check()) + quest_manager.Process(); + } - - if(quest_timers.Check()) - quest_manager.Process(); - } - } - if (InterserverTimer.Check()) { - InterserverTimer.Start(); - database.ping(); - // AsyncLoadVariables(dbasync, &database); - entity_list.UpdateWho(); - if (worldserver.TryReconnect() && (!worldserver.Connected())) - worldserver.AsyncConnect(); - } - -#ifdef EQPROFILE -#ifdef PROFILE_DUMP_TIME - if(profile_dump_timer.Check()) { - DumpZoneProfile(); - } -#endif -#endif - } //end extra profiler block - Sleep(ZoneTimerResolution); + if (InterserverTimer.Check()) { + InterserverTimer.Start(); + database.ping(); + // AsyncLoadVariables(dbasync, &database); + entity_list.UpdateWho(); + if (worldserver.TryReconnect() && (!worldserver.Connected())) + worldserver.AsyncConnect(); + } + }); + + while(RunLoops) { + EQ::EventLoop::Get().Process(); + Sleep(1); } entity_list.Clear(); @@ -566,7 +559,6 @@ int main(int argc, char** argv) { if (zone != 0) Zone::Shutdown(true); //Fix for Linux world server problem. - eqsf.Close(); worldserver.Disconnect(); safe_delete(taskmanager); command_deinit();