Merge pull request #3002 from wangyix/master

added code to nack and re-send dropped octree data packets

assignment-client/src/models/ModelServer.cpp

@@ -106,7 +106,7 @@ int ModelServer::sendSpecialPacket(OctreeQueryNode* queryNode, const SharedNodeP
             //qDebug() << "sending PacketType_MODEL_ERASE packetLength:" << packetLength;
 
             NodeList::getInstance()->writeDatagram((char*) outputBuffer, packetLength, SharedNodePointer(node));
-            queryNode->incrementSequenceNumber();
+            queryNode->packetSent(outputBuffer, packetLength);
         }
 
         nodeData->setLastDeletedModelsSentAt(deletePacketSentAt);

assignment-client/src/octree/OctreeQueryNode.cpp

@@ -41,7 +41,8 @@ OctreeQueryNode::OctreeQueryNode() :
     _sequenceNumber(0),
     _lastRootTimestamp(0),
     _myPacketType(PacketTypeUnknown),
-    _isShuttingDown(false)
+    _isShuttingDown(false),
+    _sentPacketHistory(1000)
 {
 }
 
@@ -362,3 +363,45 @@ void OctreeQueryNode::dumpOutOfView() {
         }
     }
 }
+
+void OctreeQueryNode::octreePacketSent() {
+    packetSent(_octreePacket, getPacketLength());
+}
+
+void OctreeQueryNode::packetSent(unsigned char* packet, int packetLength) {
+    packetSent(QByteArray((char*)packet, packetLength));
+}
+
+void OctreeQueryNode::packetSent(const QByteArray& packet) {
+    _sentPacketHistory.packetSent(_sequenceNumber, packet);
+    _sequenceNumber++;
+}
+
+bool OctreeQueryNode::hasNextNackedPacket() const {
+    return !_nackedSequenceNumbers.isEmpty();
+}
+
+const QByteArray* OctreeQueryNode::getNextNackedPacket() {
+    if (!_nackedSequenceNumbers.isEmpty()) {
+        // could return null if packet is not in the history
+        return _sentPacketHistory.getPacket(_nackedSequenceNumbers.takeFirst());
+    }
+    return NULL;
+}
+
+void OctreeQueryNode::parseNackPacket(QByteArray& packet) {
+
+    int numBytesPacketHeader = numBytesForPacketHeader(packet);
+    const unsigned char* dataAt = reinterpret_cast<const unsigned char*>(packet.data()) + numBytesPacketHeader;
+
+    // read number of sequence numbers
+    uint16_t numSequenceNumbers = (*(uint16_t*)dataAt);
+    dataAt += sizeof(uint16_t);
+
+    // read sequence numbers
+    for (int i = 0; i < numSequenceNumbers; i++) {
+        OCTREE_PACKET_SEQUENCE sequenceNumber = (*(OCTREE_PACKET_SEQUENCE*)dataAt);
+        _nackedSequenceNumbers.enqueue(sequenceNumber);
+        dataAt += sizeof(OCTREE_PACKET_SEQUENCE);
+    }
+}

assignment-client/src/octree/OctreeQueryNode.h

@@ -23,6 +23,8 @@
 #include <OctreeQuery.h>
 #include <OctreeSceneStats.h>
 #include <ThreadedAssignment.h> // for SharedAssignmentPointer
+#include "SentPacketHistory.h"
+#include <qqueue.h>
 
 class OctreeSendThread;
 
@@ -100,10 +102,16 @@ public:
     void forceNodeShutdown();
     bool isShuttingDown() const { return _isShuttingDown; }
 
-    void incrementSequenceNumber() { _sequenceNumber++; }
+    void octreePacketSent();
+    void packetSent(unsigned char* packet, int packetLength);
+    void packetSent(const QByteArray& packet);
 
     OCTREE_PACKET_SEQUENCE getSequenceNumber() const { return _sequenceNumber; }
-    
+
+    void parseNackPacket(QByteArray& packet);
+    bool hasNextNackedPacket() const;
+    const QByteArray* getNextNackedPacket();
+
 private slots:
     void sendThreadFinished();
     
@@ -146,6 +154,9 @@ private:
     
     PacketType _myPacketType;
     bool _isShuttingDown;
+
+    SentPacketHistory _sentPacketHistory;
+    QQueue<OCTREE_PACKET_SEQUENCE> _nackedSequenceNumbers;
 };
 
 #endif // hifi_OctreeQueryNode_h

assignment-client/src/octree/OctreeSendThread.cpp

@@ -85,6 +85,7 @@ bool OctreeSendThread::process() {
             if (nodeData && !nodeData->isShuttingDown()) {
                 bool viewFrustumChanged = nodeData->updateCurrentViewFrustum();
                 packetDistributor(nodeData, viewFrustumChanged);
+                resendNackedPackets(nodeData);
             }
         }
     }
@@ -214,8 +215,7 @@ int OctreeSendThread::handlePacketSend(OctreeQueryNode* nodeData, int& trueBytes
             packetsSent++;
 
             OctreeServer::didCallWriteDatagram(this);
-            NodeList::getInstance()->writeDatagram((char*) nodeData->getPacket(), nodeData->getPacketLength(), _node);
+            NodeList::getInstance()->writeDatagram((char*)nodeData->getPacket(), nodeData->getPacketLength(), _node);
-
             packetSent = true;
 
             thisWastedBytes = MAX_PACKET_SIZE - nodeData->getPacketLength();
@@ -244,7 +244,7 @@ int OctreeSendThread::handlePacketSend(OctreeQueryNode* nodeData, int& trueBytes
         if (nodeData->isPacketWaiting() && !nodeData->isShuttingDown()) {
             // just send the voxel packet
             OctreeServer::didCallWriteDatagram(this);
-            NodeList::getInstance()->writeDatagram((char*) nodeData->getPacket(), nodeData->getPacketLength(), _node);
+            NodeList::getInstance()->writeDatagram((char*)nodeData->getPacket(), nodeData->getPacketLength(), _node);
             packetSent = true;
 
             int thisWastedBytes = MAX_PACKET_SIZE - nodeData->getPacketLength();
@@ -274,13 +274,33 @@ int OctreeSendThread::handlePacketSend(OctreeQueryNode* nodeData, int& trueBytes
         trueBytesSent += nodeData->getPacketLength();
         truePacketsSent++;
         packetsSent++;
-        nodeData->incrementSequenceNumber();
+        nodeData->octreePacketSent();
         nodeData->resetOctreePacket();
     }
 
     return packetsSent;
 }
 
+int OctreeSendThread::resendNackedPackets(OctreeQueryNode* nodeData) {
+
+    const int MAX_PACKETS_RESEND = 10;
+    int packetsSent = 0;
+
+    const QByteArray* packet;
+    while (nodeData->hasNextNackedPacket() && packetsSent < MAX_PACKETS_RESEND) {
+        packet = nodeData->getNextNackedPacket();
+        if (packet) {
+            NodeList::getInstance()->writeDatagram(*packet, _node);
+            packetsSent++;
+
+            _totalBytes += packet->size();
+            _totalPackets++;
+            _totalWastedBytes += MAX_PACKET_SIZE - packet->size();
+        }
+    }
+    return packetsSent;
+}
+
 /// Version of voxel distributor that sends the deepest LOD level at once
 int OctreeSendThread::packetDistributor(OctreeQueryNode* nodeData, bool viewFrustumChanged) {
         

assignment-client/src/octree/OctreeSendThread.h

@@ -55,6 +55,9 @@ private:
     
     int _nodeMissingCount;
     bool _isShuttingDown;
+
+    int resendNackedPackets(OctreeQueryNode* nodeData);
+
 };
 
 #endif // hifi_OctreeSendThread_h

assignment-client/src/octree/OctreeServer.cpp

@@ -832,14 +832,13 @@ void OctreeServer::readPendingDatagrams() {
             PacketType packetType = packetTypeForPacket(receivedPacket);
             SharedNodePointer matchingNode = nodeList->sendingNodeForPacket(receivedPacket);
             if (packetType == getMyQueryMessageType()) {
-            
                 // If we got a query packet, then we're talking to an agent, and we
                 // need to make sure we have it in our nodeList.
                 if (matchingNode) {
                     nodeList->updateNodeWithDataFromPacket(matchingNode, receivedPacket);
-                    OctreeQueryNode* nodeData = (OctreeQueryNode*) matchingNode->getLinkedData();
+                    OctreeQueryNode* nodeData = (OctreeQueryNode*)matchingNode->getLinkedData();
                     if (nodeData && !nodeData->isOctreeSendThreadInitalized()) {
-                    
+                        
                         // NOTE: this is an important aspect of the proper ref counting. The send threads/node data need to 
                         // know that the OctreeServer/Assignment will not get deleted on it while it's still active. The 
                         // solution is to get the shared pointer for the current assignment. We need to make sure this is the 
@@ -848,6 +847,16 @@ void OctreeServer::readPendingDatagrams() {
                         nodeData->initializeOctreeSendThread(sharedAssignment, matchingNode);
                     }
                 }
+            } else if (packetType == PacketTypeOctreeDataNack) {
+                // If we got a nack packet, then we're talking to an agent, and we
+                // need to make sure we have it in our nodeList.
+                if (matchingNode) {
+                    nodeList->updateNodeWithDataFromPacket(matchingNode, receivedPacket);
+                    OctreeQueryNode* nodeData = (OctreeQueryNode*)matchingNode->getLinkedData();
+                    if (nodeData) {
+                        nodeData->parseNackPacket(receivedPacket);
+                    }
+                }
             } else if (packetType == PacketTypeJurisdictionRequest) {
                 _jurisdictionSender->queueReceivedPacket(matchingNode, receivedPacket);
             } else if (_octreeInboundPacketProcessor && getOctree()->handlesEditPacketType(packetType)) {

assignment-client/src/octree/SentPacketHistory.cpp

@@ -0,0 +1,44 @@
+//
+//  SentPacketHistory.cpp
+//  assignement-client/src/octree
+//
+//  Created by Yixin Wang on 6/5/2014
+//
+//  Distributed under the Apache License, Version 2.0.
+//  See the accompanying file LICENSE or http://www.apache.org/licenses/LICENSE-2.0.html
+//
+
+#include "SentPacketHistory.h"
+
+SentPacketHistory::SentPacketHistory(int size)
+    : _sentPackets(size),
+    _newestPacketAt(0),
+    _numExistingPackets(0),
+    _newestSequenceNumber(0)
+{
+}
+
+void SentPacketHistory::packetSent(OCTREE_PACKET_SEQUENCE sequenceNumber, const QByteArray& packet) {
+    _newestSequenceNumber = sequenceNumber;
+
+    // increment _newestPacketAt cyclically, insert new packet there.
+    // this will overwrite the oldest packet in the buffer
+    _newestPacketAt = (_newestPacketAt == _sentPackets.size() - 1) ? 0 : _newestPacketAt + 1;
+    _sentPackets[_newestPacketAt] = packet;
+    if (_numExistingPackets < _sentPackets.size()) {
+        _numExistingPackets++;
+    }
+}
+
+
+const QByteArray* SentPacketHistory::getPacket(OCTREE_PACKET_SEQUENCE sequenceNumber) const {
+    OCTREE_PACKET_SEQUENCE seqDiff = _newestSequenceNumber - sequenceNumber;
+    if (!(seqDiff >= 0 && seqDiff < _numExistingPackets)) {
+        return NULL;
+    }
+    int packetAt = _newestPacketAt - seqDiff;
+    if (packetAt < 0) {
+        packetAt += _sentPackets.size();
+    }
+    return &_sentPackets.at(packetAt);
+}

assignment-client/src/octree/SentPacketHistory.h

@@ -0,0 +1,35 @@
+//
+//  SentPacketHistory.h
+//  assignement-client/src/octree
+//
+//  Created by Yixin Wang on 6/5/2014
+//
+//  Distributed under the Apache License, Version 2.0.
+//  See the accompanying file LICENSE or http://www.apache.org/licenses/LICENSE-2.0.html
+//
+
+#ifndef hifi_SentPacketHistory_h
+#define hifi_SentPacketHistory_h
+
+#include <qbytearray.h>
+#include <qvector.h>
+
+#include "OctreePacketData.h"
+
+class SentPacketHistory {
+
+public:
+    SentPacketHistory(int size);
+
+    void packetSent(OCTREE_PACKET_SEQUENCE sequenceNumber, const QByteArray& packet);
+    const QByteArray* getPacket(OCTREE_PACKET_SEQUENCE sequenceNumber) const;
+
+private:
+    QVector<QByteArray> _sentPackets;    // circular buffer
+    int _newestPacketAt;
+    int _numExistingPackets;
+
+    OCTREE_PACKET_SEQUENCE _newestSequenceNumber;
+};
+
+#endif

assignment-client/src/particles/ParticleServer.cpp

@@ -106,7 +106,7 @@ int ParticleServer::sendSpecialPacket(OctreeQueryNode* queryNode, const SharedNo
             //qDebug() << "sending PacketType_PARTICLE_ERASE packetLength:" << packetLength;
 
             NodeList::getInstance()->writeDatagram((char*) outputBuffer, packetLength, SharedNodePointer(node));
-            queryNode->incrementSequenceNumber();
+            queryNode->packetSent(outputBuffer, packetLength);
         }
 
         nodeData->setLastDeletedParticlesSentAt(deletePacketSentAt);

assignment-client/src/voxels/VoxelServer.cpp

@@ -75,7 +75,7 @@ int VoxelServer::sendSpecialPacket(OctreeQueryNode* queryNode, const SharedNodeP
     }
 
     NodeList::getInstance()->writeDatagram((char*) _tempOutputBuffer, envPacketLength, SharedNodePointer(node));
-    queryNode->incrementSequenceNumber();
+    queryNode->packetSent(_tempOutputBuffer, envPacketLength);
 
     return envPacketLength;
 }

interface/src/Application.cpp

@@ -167,7 +167,8 @@ Application::Application(int& argc, char** argv, QElapsedTimer &startup_time) :
         _applicationOverlay(),
         _runningScriptsWidget(new RunningScriptsWidget(_window)),
         _runningScriptsWidgetWasVisible(false),
-        _trayIcon(new QSystemTrayIcon(_window))
+        _trayIcon(new QSystemTrayIcon(_window)),
+        _lastNackTime(usecTimestampNow())
 {
     // read the ApplicationInfo.ini file for Name/Version/Domain information
     QSettings applicationInfo(Application::resourcesPath() + "info/ApplicationInfo.ini", QSettings::IniFormat);
@@ -2093,6 +2094,78 @@ void Application::updateMyAvatar(float deltaTime) {
             _lastQueriedViewFrustum = _viewFrustum;
         }
     }
+
+    // sent a nack packet containing missing sequence numbers of received packets
+    {
+        quint64 now = usecTimestampNow();
+        quint64 sinceLastNack = now - _lastNackTime;
+        const quint64 TOO_LONG_SINCE_LAST_NACK = 250 * MSECS_PER_SECOND;
+        if (sinceLastNack > TOO_LONG_SINCE_LAST_NACK) {
+            _lastNackTime = now;
+            sendNack();
+        }
+    }
+}
+
+void Application::sendNack() {
+
+    char packet[MAX_PACKET_SIZE];
+    NodeList* nodeList = NodeList::getInstance();
+
+    // iterates thru all nodes in NodeList
+    foreach(const SharedNodePointer& node, NodeList::getInstance()->getNodeHash()) {
+        
+        if (node->getActiveSocket() &&
+            ( node->getType() == NodeType::VoxelServer
+            || node->getType() == NodeType::ParticleServer
+            || node->getType() == NodeType::ModelServer)
+            ) {
+
+            QUuid nodeUUID = node->getUUID();
+
+            _octreeSceneStatsLock.lockForRead();
+
+            // retreive octree scene stats of this node
+            if (_octreeServerSceneStats.find(nodeUUID) == _octreeServerSceneStats.end()) {
+                _octreeSceneStatsLock.unlock();
+                continue;
+            }
+            OctreeSceneStats& stats = _octreeServerSceneStats[nodeUUID];
+
+            // check if there are any sequence numbers that need to be nacked
+            int numSequenceNumbersAvailable = stats.getNumSequenceNumbersToNack();
+            if (numSequenceNumbersAvailable == 0) {
+                _octreeSceneStatsLock.unlock();
+                continue;
+            }
+
+            char* dataAt = packet;
+            int bytesRemaining = MAX_PACKET_SIZE;
+
+            // pack header
+            int numBytesPacketHeader = populatePacketHeader(packet, PacketTypeOctreeDataNack);
+            dataAt += numBytesPacketHeader;
+            bytesRemaining -= numBytesPacketHeader;
+            int numSequenceNumbersRoomFor = (bytesRemaining - sizeof(uint16_t)) / sizeof(OCTREE_PACKET_SEQUENCE);
+
+            // calculate and pack the number of sequence numbers
+            uint16_t numSequenceNumbers = min(numSequenceNumbersAvailable, numSequenceNumbersRoomFor);
+            uint16_t* numSequenceNumbersAt = (uint16_t*)dataAt;
+            *numSequenceNumbersAt = numSequenceNumbers;
+            dataAt += sizeof(uint16_t);
+
+            // pack sequence numbers
+            for (int i = 0; i < numSequenceNumbers; i++) {
+                OCTREE_PACKET_SEQUENCE* sequenceNumberAt = (OCTREE_PACKET_SEQUENCE*)dataAt;
+                *sequenceNumberAt = stats.getNextSequenceNumberToNack();
+                dataAt += sizeof(OCTREE_PACKET_SEQUENCE);
+            }
+            
+            _octreeSceneStatsLock.unlock();
+
+            nodeList->writeUnverifiedDatagram(packet, dataAt - packet, node);
+        }
+    }
 }
 
 void Application::queryOctree(NodeType_t serverType, PacketType packetType, NodeToJurisdictionMap& jurisdictions) {

interface/src/Application.h

@@ -411,6 +411,8 @@ private:
     static void attachNewHeadToNode(Node *newNode);
     static void* networkReceive(void* args); // network receive thread
 
+    void sendNack();
+
     MainWindow* _window;
     GLCanvas* _glWidget; // our GLCanvas has a couple extra features
 
@@ -580,6 +582,8 @@ private:
     bool _runningScriptsWidgetWasVisible;
 
     QSystemTrayIcon* _trayIcon;
+
+    quint64 _lastNackTime;
 };
 
 #endif // hifi_Application_h

libraries/networking/src/PacketHeaders.cpp

@@ -108,7 +108,7 @@ int populatePacketHeader(char* packet, PacketType type, const QUuid& connectionU
     position += NUM_BYTES_RFC4122_UUID;
     
     if (!NON_VERIFIED_PACKETS.contains(type)) {
-        // pack 16 bytes of zeros where the md5 hash will be placed one data is packed
+        // pack 16 bytes of zeros where the md5 hash will be placed once data is packed
         memset(position, 0, NUM_BYTES_MD5_HASH);
         position += NUM_BYTES_MD5_HASH;
     }

libraries/networking/src/PacketHeaders.h

@@ -66,6 +66,7 @@ enum PacketType {
     PacketTypeModelAddOrEdit,
     PacketTypeModelErase,
     PacketTypeModelAddResponse,
+    PacketTypeOctreeDataNack
 };
 
 typedef char PacketVersion;
@@ -74,7 +75,8 @@ const QSet<PacketType> NON_VERIFIED_PACKETS = QSet<PacketType>()
     << PacketTypeDomainServerRequireDTLS << PacketTypeDomainConnectRequest
     << PacketTypeDomainList << PacketTypeDomainListRequest << PacketTypeDomainOAuthRequest
     << PacketTypeCreateAssignment << PacketTypeRequestAssignment << PacketTypeStunResponse
-    << PacketTypeNodeJsonStats << PacketTypeVoxelQuery << PacketTypeParticleQuery << PacketTypeModelQuery;
+    << PacketTypeNodeJsonStats << PacketTypeVoxelQuery << PacketTypeParticleQuery << PacketTypeModelQuery
+    << PacketTypeOctreeDataNack;
 
 const int NUM_BYTES_MD5_HASH = 16;
 const int NUM_STATIC_HEADER_BYTES = sizeof(PacketVersion) + NUM_BYTES_RFC4122_UUID;

libraries/octree/src/OctreeSceneStats.cpp

@@ -46,6 +46,7 @@ OctreeSceneStats::OctreeSceneStats() :
     _incomingReallyLate(0),
     _incomingPossibleDuplicate(0),
     _missingSequenceNumbers(),
+    _sequenceNumbersToNack(),
     _incomingFlightTimeAverage(samples),
     _jurisdictionRoot(NULL)
 {
@@ -158,6 +159,7 @@ void OctreeSceneStats::copyFromOther(const OctreeSceneStats& other) {
     _incomingPossibleDuplicate = other._incomingPossibleDuplicate;
     
     _missingSequenceNumbers = other._missingSequenceNumbers;
+    _sequenceNumbersToNack = other._sequenceNumbersToNack;
 }
 
 
@@ -926,6 +928,7 @@ void OctreeSceneStats::trackIncomingOctreePacket(const QByteArray& packet,
                         qDebug() << "found it in _missingSequenceNumbers";
                     }
                     _missingSequenceNumbers.remove(sequence);
+                    _sequenceNumbersToNack.remove(sequence);
                     _incomingLikelyLost--;
                     _incomingRecovered++;
                 } else {
@@ -955,6 +958,7 @@ void OctreeSceneStats::trackIncomingOctreePacket(const QByteArray& packet,
                 _incomingLikelyLost += missing;
                 for(unsigned int missingSequence = expected; missingSequence < sequence; missingSequence++) {
                     _missingSequenceNumbers << missingSequence;
+                    _sequenceNumbersToNack << missingSequence;
                 }
             }
         }
@@ -982,9 +986,20 @@ void OctreeSceneStats::trackIncomingOctreePacket(const QByteArray& packet,
                     qDebug() << "pruning really old missing sequence:" << missingItem;
                 }
                 _missingSequenceNumbers.remove(missingItem);
+                _sequenceNumbersToNack.remove(missingItem);
             }
         }
     }
 
 }
 
+int OctreeSceneStats::getNumSequenceNumbersToNack() const {
+    return _sequenceNumbersToNack.size();
+}
+
+uint16_t OctreeSceneStats::getNextSequenceNumberToNack() {
+    QSet<uint16_t>::Iterator it = _sequenceNumbersToNack.begin();
+    uint16_t sequenceNumber = *it;
+    _sequenceNumbersToNack.remove(sequenceNumber);
+    return sequenceNumber;
+}

libraries/octree/src/OctreeSceneStats.h

@@ -16,6 +16,7 @@
 #include <NodeList.h>
 #include <SharedUtil.h>
 #include "JurisdictionMap.h"
+#include "OctreePacketData.h"
 
 #define GREENISH  0x40ff40d0
 #define YELLOWISH 0xffef40c0
@@ -172,6 +173,9 @@ public:
     quint32 getIncomingPossibleDuplicate() const { return _incomingPossibleDuplicate; }
     float getIncomingFlightTimeAverage() { return _incomingFlightTimeAverage.getAverage(); }
 
+    int getNumSequenceNumbersToNack() const;
+    OCTREE_PACKET_SEQUENCE getNextSequenceNumberToNack();
+
 private:
 
     void copyFromOther(const OctreeSceneStats& other);
@@ -272,7 +276,8 @@ private:
     quint32 _incomingLate; /// out of order later than expected
     quint32 _incomingReallyLate; /// out of order and later than MAX_MISSING_SEQUENCE_OLD_AGE late
     quint32 _incomingPossibleDuplicate; /// out of order possibly a duplicate
-    QSet<uint16_t> _missingSequenceNumbers;
+    QSet<OCTREE_PACKET_SEQUENCE> _missingSequenceNumbers;
+    QSet<OCTREE_PACKET_SEQUENCE> _sequenceNumbersToNack;
     SimpleMovingAverage _incomingFlightTimeAverage;
     
     // features related items