From 3209e361761f2e9941095d43f31609fcd06d78ec Mon Sep 17 00:00:00 2001 From: Jocke Date: Wed, 9 Mar 2016 18:26:33 +0100 Subject: [PATCH] We can now send more data with UDP than previously, the splitting seems to work fine but receiving has some problems. --- include/Engine/Network/Packet.h | 3 +- include/Engine/Network/PlayerDefinition.h | 1 + include/Engine/Network/UDPClient.h | 10 +- include/Engine/Network/UDPServer.h | 2 +- src/Engine/Network/Client.cpp | 14 ++- src/Engine/Network/Packet.cpp | 24 +++- src/Engine/Network/Server.cpp | 1 + src/Engine/Network/UDPClient.cpp | 129 ++++++++++++++++++---- src/Engine/Network/UDPServer.cpp | 31 +++--- 9 files changed, 161 insertions(+), 54 deletions(-) diff --git a/include/Engine/Network/Packet.h b/include/Engine/Network/Packet.h index 6095e6a6..11a1a4bf 100644 --- a/include/Engine/Network/Packet.h +++ b/include/Engine/Network/Packet.h @@ -59,10 +59,11 @@ public: void UpdateSize(); char* ReadData(int SizeOfData); void ChangePacketID(unsigned int& packetID); - void ChangeSequenceNumber(int sequenceNumber, int sequenceLength); + void ChangeSequenceNumber(int sequenceNumber, int sequenceLength, int groupNumber); size_t Size() { return m_Offset; }; char* Data() { return m_Data; }; MessageType GetMessageType(); + size_t Group(); size_t DataReadSize() { return m_ReturnDataOffset; } size_t MaxSize() { return m_MaxPacketSize; } size_t HeaderSize() { return m_HeaderSize; } diff --git a/include/Engine/Network/PlayerDefinition.h b/include/Engine/Network/PlayerDefinition.h index afd5d889..6ccd1034 100644 --- a/include/Engine/Network/PlayerDefinition.h +++ b/include/Engine/Network/PlayerDefinition.h @@ -14,6 +14,7 @@ struct PlayerDefinition { unsigned short TCPPort; // use for tcp connections boost::shared_ptr TCPSocket; + int PacketGroup = 1; }; #endif diff --git a/include/Engine/Network/UDPClient.h b/include/Engine/Network/UDPClient.h index 94b23080..b2b156e4 100644 --- a/include/Engine/Network/UDPClient.h +++ b/include/Engine/Network/UDPClient.h @@ -1,6 +1,7 @@ #ifndef UDPClient_h__ #define UDPClient_h__ #include +#include #include #include "Network/NetworkClient.h" @@ -15,18 +16,23 @@ public: void Connect(std::string playerName, std::string address, int port); void Disconnect(); void Receive(Packet& packet); - void Send(Packet & packet); + void ReceivePackets(); + void Send(Packet& packet); void Broadcast(Packet& packet, int port); bool IsSocketAvailable(); + // Returns false if no packets are available + bool GetNextPacket(Packet& packet); private: // Assio UDP logic boost::asio::io_service m_IOService; boost::asio::ip::udp::endpoint m_ReceiverEndpoint; boost::shared_ptr m_Socket; + int lastReceivedSnapshotGroup = 0; int readBuffer(); PacketID m_SendPacketID = 0; //map:(packetGroup, vector:(pair:(groupIndex, packetData))) - std::map>>> packetSegmentMap; + std::map>>> m_PacketSegmentMap; + bool hasReceivedPacket(int packetGroup, int groupIndex); }; #endif \ No newline at end of file diff --git a/include/Engine/Network/UDPServer.h b/include/Engine/Network/UDPServer.h index 6b6b7d33..1d37f642 100644 --- a/include/Engine/Network/UDPServer.h +++ b/include/Engine/Network/UDPServer.h @@ -3,7 +3,7 @@ #include "NetworkServer.h" #include -#define MAXPACKETSIZE 32000 +#define MAXPACKETSIZE 64000 class UDPServer : public NetworkServer { diff --git a/src/Engine/Network/Client.cpp b/src/Engine/Network/Client.cpp index 2bda83db..9c494cae 100644 --- a/src/Engine/Network/Client.cpp +++ b/src/Engine/Network/Client.cpp @@ -50,13 +50,15 @@ void Client::Update() { m_EventBroker->Process(); while (m_Unreliable.IsSocketAvailable()) { - // Packet will get real data in receive - Packet packet(MessageType::Invalid); - m_Unreliable.Receive(packet); - if (packet.GetMessageType() == MessageType::Connect) { - parseUDPConnect(packet); + m_Unreliable.ReceivePackets(); + } + // Packet will get real data in GetNextPacket() + Packet parsedPacket(MessageType::Invalid); + while (m_Unreliable.GetNextPacket(parsedPacket)) { + if (parsedPacket.GetMessageType() == MessageType::Connect) { + parseUDPConnect(parsedPacket); } else { - parseMessageType(packet); + parseMessageType(parsedPacket); } } diff --git a/src/Engine/Network/Packet.cpp b/src/Engine/Network/Packet.cpp index 75b69bd7..3e4bf043 100644 --- a/src/Engine/Network/Packet.cpp +++ b/src/Engine/Network/Packet.cpp @@ -3,16 +3,22 @@ Packet::Packet(MessageType type, unsigned int& packetID) { m_Data = new char[m_MaxPacketSize]; - Init(type, packetID, 1, 1 , -1); + Init(type, packetID, 1, 1, -1); } // Create message Packet::Packet(char* data, const size_t sizeOfPacket) { + // Create message header + // allocate memory for size of packet, sequenceNumber and totalPacketesInSequence + m_ReturnDataOffset = 0; + m_Offset = 0; // Resize message m_MaxPacketSize = sizeOfPacket; // Copy data newly allocated memory m_Data = new char[sizeOfPacket]; + unsigned int dummy = 0; + Init(MessageType::Invalid, dummy, 0, 0, 0); memcpy(m_Data, data, sizeOfPacket); m_Offset = sizeOfPacket; } @@ -119,7 +125,7 @@ void Packet::ReconstructFromData(char * data, size_t sizeOfData) void Packet::UpdateSize() { int whatisoffset = m_Offset; - memcpy(m_Data, &m_Offset, sizeof(int)); + memcpy(m_Data + packetSizeOffset, &m_Offset, sizeof(int)); } char * Packet::ReadData(int sizeOfData) @@ -140,10 +146,11 @@ void Packet::ChangePacketID(unsigned int & packetID) memcpy(m_Data + packetIDOffset, &packetID, sizeof(int)); } -void Packet::ChangeSequenceNumber(int groupIndex, int groupSize) +void Packet::ChangeSequenceNumber(int groupIndex, int groupSize, int groupNumber) { - memcpy(&groupIndex, m_Data + groupIndexOffset, sizeof(int)); - memcpy(&groupSize, m_Data + groupSizeOffset, sizeof(int)); + memcpy(m_Data + groupIndexOffset, &groupIndex, sizeof(int)); + memcpy(m_Data + groupSizeOffset, &groupSize, sizeof(int)); + memcpy(m_Data + groupOffset, &groupNumber, sizeof(int)); } MessageType Packet::GetMessageType() @@ -153,6 +160,13 @@ MessageType Packet::GetMessageType() return messagType; } +size_t Packet::Group() +{ + size_t groupNumber; + memcpy(&groupNumber, m_Data + groupOffset, sizeof(int)); + return groupNumber; +} + size_t Packet::GroupIndex() { size_t groupIndex; diff --git a/src/Engine/Network/Server.cpp b/src/Engine/Network/Server.cpp index 7a2d58a8..f81e0443 100644 --- a/src/Engine/Network/Server.cpp +++ b/src/Engine/Network/Server.cpp @@ -171,6 +171,7 @@ void Server::sendSnapshot() Packet packet(MessageType::Snapshot); addInputCommandsToPacket(packet); addPlayersToPacket(packet, EntityID_Invalid); + //addChildrenToPacket(packet, EntityID_Invalid); unreliableBroadcast(packet); } diff --git a/src/Engine/Network/UDPClient.cpp b/src/Engine/Network/UDPClient.cpp index 30690757..dbc4d979 100644 --- a/src/Engine/Network/UDPClient.cpp +++ b/src/Engine/Network/UDPClient.cpp @@ -3,12 +3,10 @@ using namespace boost::asio::ip; UDPClient::UDPClient() -{ -} +{ } UDPClient::~UDPClient() -{ -} +{ } void UDPClient::Connect(std::string playerName, std::string address, int port) { @@ -27,12 +25,18 @@ void UDPClient::Disconnect() void UDPClient::Receive(Packet& packet) { - int bytesRead = readBuffer(); - if (bytesRead > 0) { - packet.ReconstructFromData(m_ReadBuffer, bytesRead); - } + // int bytesRead = readBuffer(); + //if (bytesRead > 0) { + // packet.ReconstructFromData(m_ReadBuffer, bytesRead); + //} } +void UDPClient::ReceivePackets() +{ + readBuffer(); +} + + int UDPClient::readBuffer() { if (!m_Socket) { @@ -40,36 +44,67 @@ int UDPClient::readBuffer() } boost::system::error_code error; // Peek header - m_Socket->receive(boost + m_Socket->receive(boost ::asio::buffer((void*)m_ReadBuffer, 5 * sizeof(int)), - boost::asio::ip::udp::socket::message_peek, error); + boost::asio::ip::udp::socket::message_peek, error); + int sizeOfPacket = 0; memcpy(&sizeOfPacket, m_ReadBuffer, sizeof(int)); - int packetID = 0; - memcpy(&packetID, m_ReadBuffer + 4 * sizeof(int), sizeof(int)); + if (sizeOfPacket == 0) { + return 0; + } + int packetGroup = 0; + memcpy(&packetGroup, m_ReadBuffer + sizeof(int), sizeof(int)); + int packetGroupIndex = 0; + memcpy(&packetGroupIndex, m_ReadBuffer + 2 * sizeof(int), sizeof(int)); + int packetGroupSize = 0; + memcpy(&packetGroupSize, m_ReadBuffer + 3 * sizeof(int), sizeof(int)); + //LOG_INFO("Packet group: %i. Group index: %i. Group size: %i", packetGroup, packetGroupIndex, packetGroupSize); + //LOG_INFO("Packet size: %i.", sizeOfPacket); if (sizeOfPacket > m_Socket->available()) { LOG_WARNING("UDPClient::readBuffer(): We haven't got the whole packet yet."); // return; } // if the buffer is to small increase the size of it - if (sizeOfPacket > m_BufferSize) { - delete[] m_ReadBuffer; - m_ReadBuffer = new char[sizeOfPacket]; - m_BufferSize = sizeOfPacket; - } + boost::shared_ptr packetData(new char[sizeOfPacket]); + // Read the message size_t bytesReceived = m_Socket->receive_from(boost - ::asio::buffer((void*)(m_ReadBuffer), + ::asio::buffer((void*)(packetData.get()), sizeOfPacket), m_ReceiverEndpoint, 0, error); if (error) { - //LOG_ERROR("receive: %s", error.message().c_str()); + LOG_ERROR("receive: %s", error.message().c_str()); + } + // Might want to do this earlier when i figure out a good way to + // remove data from network buffer. + if (hasReceivedPacket(packetGroup, packetGroupIndex)) { + return 0; + } + //std::map>>> packetSegmentMap; + // If group exists + if (m_PacketSegmentMap.find(packetGroup) != m_PacketSegmentMap.end()) { + m_PacketSegmentMap.at(packetGroup).push_back(std::make_pair(packetGroupIndex, std::move(packetData))); + } else { // Create group and add element + m_PacketSegmentMap[packetGroup].push_back(std::make_pair(packetGroupIndex, std::move(packetData))); } - // new char [sizeOfPacket] save in map with packetID as key return bytesReceived; } +bool UDPClient::hasReceivedPacket(int packetGroup, int groupIndex) +{ + if (m_PacketSegmentMap.find(packetGroup) != m_PacketSegmentMap.end()) { + const std::vector>>& loopPacketGroup = m_PacketSegmentMap.at(packetGroup); + for (size_t i = 0; i < loopPacketGroup.size(); i++) { + if (loopPacketGroup.at(i).first == groupIndex) { + return true; + } + } + } + return false; +} + void UDPClient::Send(Packet& packet) { packet.UpdateSize(); @@ -77,7 +112,7 @@ void UDPClient::Send(Packet& packet) packet.Data(), packet.Size()), m_ReceiverEndpoint, 0); -} +} void UDPClient::Broadcast(Packet& packet, int port) { @@ -93,8 +128,56 @@ void UDPClient::Broadcast(Packet& packet, int port) bool UDPClient::IsSocketAvailable() { - if (!m_Socket) { + if (!m_Socket) { return false; } return m_Socket->available(); -} \ No newline at end of file +} + +bool UDPClient::GetNextPacket(Packet & packet) +{ + // A duplicate packet should not be present in the vector! + // Soo we will assume that this is true and only look if size + // of vector is correct. + std::map>>>::iterator it = m_PacketSegmentMap.begin(); + while (it != m_PacketSegmentMap.end()) { + // pair(Group index, packetData) + std::vector>>& currentVector = it->second; + Packet headerInfoPacket(currentVector.at(0).second.get(), packet.HeaderSize()); + int groupSize = headerInfoPacket.GroupSize(); + //LOG_INFO("UDPClient::GetNextPacket: Packet group : %i.Group index : %i.Group size : %i. lastReceivedSnapshotGroup: %i. MessageType(Ples 4): %i", headerInfoPacket.Group(), headerInfoPacket.GroupIndex(), groupSize, lastReceivedSnapshotGroup, headerInfoPacket.GetMessageType()); + //LOG_INFO("UDPClient::GetNextPacket: Packet group : %i.lastReceivedSnapshotGroup: %i. MessageType(Ples 4): %i", headerInfoPacket.Group(), lastReceivedSnapshotGroup, headerInfoPacket.GetMessageType()); + int mapSize = m_PacketSegmentMap.size(); + if (mapSize > 5) { + it = m_PacketSegmentMap.erase(it); + LOG_INFO("The map is increasing in size, size is %i", mapSize); + continue; + } + if (headerInfoPacket.GetMessageType() == MessageType::Snapshot && lastReceivedSnapshotGroup > headerInfoPacket.Group()) { + it = m_PacketSegmentMap.erase(it); + continue; + //LOG_INFO("Deleted old entry"); + } + if (currentVector.size() == groupSize) { + std::sort(currentVector.begin(), currentVector.end()); + // Add the first packet in vector + packet.ReconstructFromData(currentVector.at(0).second.get(), packet.HeaderSize()); + // Add the rest of the packets. + int sizeOfData = 0; + for (size_t i = 0; i < currentVector.size(); i++) { + memcpy(&sizeOfData, currentVector.at(i).second.get(), sizeof(int)); + packet.WriteData(currentVector.at(i).second.get() + packet.HeaderSize(), sizeOfData - packet.HeaderSize()); + } + if (headerInfoPacket.GetMessageType() == MessageType::Snapshot) { + lastReceivedSnapshotGroup = packet.Group(); + } + // No need to get next it as we are returning. + //LOG_INFO("Packet parsed"); + m_PacketSegmentMap.erase(it); + return true; + } else { + ++it; + } + } + return false; +} diff --git a/src/Engine/Network/UDPServer.cpp b/src/Engine/Network/UDPServer.cpp index 6ed95fe0..cf6844d9 100644 --- a/src/Engine/Network/UDPServer.cpp +++ b/src/Engine/Network/UDPServer.cpp @@ -12,22 +12,25 @@ UDPServer::UDPServer(int port) UDPServer::~UDPServer() { } - +// TODO: Fix correct groups void UDPServer::Send(Packet& packet, PlayerDefinition & playerDefinition) { packet.UpdateSize(); try { + //Debug + int debugTheSixeOfpacket = packet.Size(); + //Debug end // Remove header from packet. packet.ReadData(packet.HeaderSize()); - int bytesSent = 0; + int totalBytesSent = 0; int sequenceNumber = 1; - int totalMessages = std::ceil(packet.Size() / MAXPACKETSIZE); + int totalMessages = std::ceil((float)packet.Size() / MAXPACKETSIZE); int packetDataSent = 0; int packetDataSize = packet.Size() - packet.HeaderSize(); + while (packetDataSize > packetDataSent) { - Packet splitPacket(packet.GetMessageType(), playerDefinition.PacketID); - splitPacket.ChangeSequenceNumber(sequenceNumber, totalMessages); + splitPacket.ChangeSequenceNumber(sequenceNumber, totalMessages, playerDefinition.PacketGroup); int amountToSend = packetDataSize - packetDataSent; if (amountToSend > MAXPACKETSIZE) { amountToSend = MAXPACKETSIZE; @@ -36,26 +39,22 @@ void UDPServer::Send(Packet& packet, PlayerDefinition & playerDefinition) splitPacket.UpdateSize(); // Remove header size from bytes sent soo that we only // count data in the packet - - bytesSent += m_Socket->send_to( - boost::asio::buffer(splitPacket.Data() + bytesSent, splitPacket.Size()), + int bytesSent = 0; + bytesSent = m_Socket->send_to( + boost::asio::buffer(splitPacket.Data(), splitPacket.Size()), playerDefinition.Endpoint, 0); - packetDataSent = bytesSent - splitPacket.HeaderSize(); + packetDataSent += bytesSent - splitPacket.HeaderSize(); + totalBytesSent += bytesSent; + // LOG_INFO("bytesSent size %i, groupIndex: %i. Number of packets: %i", bytesSent, sequenceNumber, totalMessages); ++sequenceNumber; - int packetsise = splitPacket.ReadPrimitive(); - int groopOffset = splitPacket.ReadPrimitive(); - int groopIndes = splitPacket.ReadPrimitive(); - int groopSiseOffset = splitPacket.ReadPrimitive(); - int mezzagetype = splitPacket.ReadPrimitive(); - int pakketIDOffset = splitPacket.ReadPrimitive(); } + playerDefinition.PacketGroup++; } catch (const boost::system::system_error& e) { LOG_INFO(e.what()); // TODO: Clean up invalid endpoints out of m_ConnectedPlayers later playerDefinition.Endpoint = boost::asio::ip::udp::endpoint(); } - } // Send back to endpoint of received packet void UDPServer::Send(Packet & packet)