moved sendNack() into packetDistributor()

...for the purpose of enforcing packets sent per interval in
OctreeSendThread.  Corrected mistake in keeping track of number of
special packets sent where sendSpecialPacket() was assumed to only send
one packet per call.
This commit is contained in:
wangyix 2014-06-10 15:35:48 -07:00
parent b96b6c9857
commit 5ab271330e
9 changed files with 34 additions and 37 deletions

View file

@ -86,7 +86,7 @@ bool ModelServer::hasSpecialPacketToSend(const SharedNodePointer& node) {
return shouldSendDeletedModels;
}
int ModelServer::sendSpecialPacket(OctreeQueryNode* queryNode, const SharedNodePointer& node) {
int ModelServer::sendSpecialPacket(const SharedNodePointer& node, OctreeQueryNode* queryNode, int& packetsSent) {
unsigned char outputBuffer[MAX_PACKET_SIZE];
size_t packetLength = 0;
@ -99,6 +99,7 @@ int ModelServer::sendSpecialPacket(OctreeQueryNode* queryNode, const SharedNodeP
bool hasMoreToSend = true;
// TODO: is it possible to send too many of these packets? what if you deleted 1,000,000 models?
packetsSent = 0;
while (hasMoreToSend) {
hasMoreToSend = tree->encodeModelsDeletedSince(queryNode->getSequenceNumber(), deletedModelsSentAt,
outputBuffer, MAX_PACKET_SIZE, packetLength);
@ -107,6 +108,7 @@ int ModelServer::sendSpecialPacket(OctreeQueryNode* queryNode, const SharedNodeP
NodeList::getInstance()->writeDatagram((char*) outputBuffer, packetLength, SharedNodePointer(node));
queryNode->packetSent(outputBuffer, packetLength);
packetsSent++;
}
nodeData->setLastDeletedModelsSentAt(deletePacketSentAt);

View file

@ -37,7 +37,7 @@ public:
// subclass may implement these method
virtual void beforeRun();
virtual bool hasSpecialPacketToSend(const SharedNodePointer& node);
virtual int sendSpecialPacket(OctreeQueryNode* queryNode, const SharedNodePointer& node);
virtual int sendSpecialPacket(const SharedNodePointer& node, OctreeQueryNode* queryNode, int& packetsSent);
virtual void modelCreated(const ModelItem& newModel, const SharedNodePointer& senderNode);

View file

@ -85,7 +85,6 @@ bool OctreeSendThread::process() {
if (nodeData && !nodeData->isShuttingDown()) {
bool viewFrustumChanged = nodeData->updateCurrentViewFrustum();
packetDistributor(nodeData, viewFrustumChanged);
resendNackedPackets(nodeData);
}
}
}
@ -281,26 +280,6 @@ int OctreeSendThread::handlePacketSend(OctreeQueryNode* nodeData, int& trueBytes
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) {
@ -311,6 +290,11 @@ int OctreeSendThread::packetDistributor(OctreeQueryNode* nodeData, bool viewFrus
return 0;
}
// calculate max number of packets that can be sent during this interval
int clientMaxPacketsPerInterval = std::max(1, (nodeData->getMaxOctreePacketsPerSecond() / INTERVALS_PER_SECOND));
int maxPacketsPerInterval = std::min(clientMaxPacketsPerInterval, _myServer->getPacketsPerClientPerInterval());
int truePacketsSent = 0;
int trueBytesSent = 0;
int packetsSentThisInterval = 0;
@ -408,9 +392,6 @@ int OctreeSendThread::packetDistributor(OctreeQueryNode* nodeData, bool viewFrus
//quint64 startCompressTimeMsecs = OctreePacketData::getCompressContentTime() / 1000;
//quint64 startCompressCalls = OctreePacketData::getCompressContentCalls();
int clientMaxPacketsPerInterval = std::max(1,(nodeData->getMaxOctreePacketsPerSecond() / INTERVALS_PER_SECOND));
int maxPacketsPerInterval = std::min(clientMaxPacketsPerInterval, _myServer->getPacketsPerClientPerInterval());
int extraPackingAttempts = 0;
bool completedScene = false;
while (somethingToSend && packetsSentThisInterval < maxPacketsPerInterval && !nodeData->isShuttingDown()) {
@ -581,12 +562,26 @@ int OctreeSendThread::packetDistributor(OctreeQueryNode* nodeData, bool viewFrus
// send the environment packet
// TODO: should we turn this into a while loop to better handle sending multiple special packets
if (_myServer->hasSpecialPacketToSend(_node) && !nodeData->isShuttingDown()) {
trueBytesSent += _myServer->sendSpecialPacket(nodeData, _node);
int specialPacketsSent;
trueBytesSent += _myServer->sendSpecialPacket(_node, nodeData, specialPacketsSent);
nodeData->resetOctreePacket(); // because nodeData's _sequenceNumber has changed
truePacketsSent++;
packetsSentThisInterval++;
truePacketsSent += specialPacketsSent;
packetsSentThisInterval += specialPacketsSent;
}
// Re-send packets that were nacked by the client
while (nodeData->hasNextNackedPacket() && packetsSentThisInterval < maxPacketsPerInterval) {
const QByteArray* packet = nodeData->getNextNackedPacket();
if (packet) {
NodeList::getInstance()->writeDatagram(*packet, _node);
truePacketsSent++;
packetsSentThisInterval++;
_totalBytes += packet->size();
_totalPackets++;
_totalWastedBytes += MAX_PACKET_SIZE - packet->size();
}
}
quint64 end = usecTimestampNow();
int elapsedmsec = (end - start)/USECS_PER_MSEC;

View file

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

View file

@ -72,7 +72,7 @@ public:
// subclass may implement these method
virtual void beforeRun() { };
virtual bool hasSpecialPacketToSend(const SharedNodePointer& node) { return false; }
virtual int sendSpecialPacket(OctreeQueryNode* queryNode, const SharedNodePointer& node) { return 0; }
virtual int sendSpecialPacket(const SharedNodePointer& node, OctreeQueryNode* queryNode, int& packetsSent) { return 0; }
static void attachQueryNodeToNode(Node* newNode);

View file

@ -86,7 +86,7 @@ bool ParticleServer::hasSpecialPacketToSend(const SharedNodePointer& node) {
return shouldSendDeletedParticles;
}
int ParticleServer::sendSpecialPacket(OctreeQueryNode* queryNode, const SharedNodePointer& node) {
int ParticleServer::sendSpecialPacket(const SharedNodePointer& node, OctreeQueryNode* queryNode, int& packetsSent) {
unsigned char outputBuffer[MAX_PACKET_SIZE];
size_t packetLength = 0;
@ -99,6 +99,7 @@ int ParticleServer::sendSpecialPacket(OctreeQueryNode* queryNode, const SharedNo
bool hasMoreToSend = true;
// TODO: is it possible to send too many of these packets? what if you deleted 1,000,000 particles?
packetsSent = 0;
while (hasMoreToSend) {
hasMoreToSend = tree->encodeParticlesDeletedSince(queryNode->getSequenceNumber(), deletedParticlesSentAt,
outputBuffer, MAX_PACKET_SIZE, packetLength);
@ -107,6 +108,7 @@ int ParticleServer::sendSpecialPacket(OctreeQueryNode* queryNode, const SharedNo
NodeList::getInstance()->writeDatagram((char*) outputBuffer, packetLength, SharedNodePointer(node));
queryNode->packetSent(outputBuffer, packetLength);
packetsSent++;
}
nodeData->setLastDeletedParticlesSentAt(deletePacketSentAt);

View file

@ -37,7 +37,7 @@ public:
// subclass may implement these method
virtual void beforeRun();
virtual bool hasSpecialPacketToSend(const SharedNodePointer& node);
virtual int sendSpecialPacket(OctreeQueryNode* queryNode, const SharedNodePointer& node);
virtual int sendSpecialPacket(const SharedNodePointer& node, OctreeQueryNode* queryNode, int& packetsSent);
virtual void particleCreated(const Particle& newParticle, const SharedNodePointer& senderNode);

View file

@ -40,7 +40,7 @@ bool VoxelServer::hasSpecialPacketToSend(const SharedNodePointer& node) {
return shouldSendEnvironments;
}
int VoxelServer::sendSpecialPacket(OctreeQueryNode* queryNode, const SharedNodePointer& node) {
int VoxelServer::sendSpecialPacket(const SharedNodePointer& node, OctreeQueryNode* queryNode, int& packetsSent) {
unsigned char* copyAt = _tempOutputBuffer;
@ -76,6 +76,7 @@ int VoxelServer::sendSpecialPacket(OctreeQueryNode* queryNode, const SharedNodeP
NodeList::getInstance()->writeDatagram((char*) _tempOutputBuffer, envPacketLength, SharedNodePointer(node));
queryNode->packetSent(_tempOutputBuffer, envPacketLength);
packetsSent = 1;
return envPacketLength;
}

View file

@ -46,7 +46,7 @@ public:
// subclass may implement these method
virtual void beforeRun();
virtual bool hasSpecialPacketToSend(const SharedNodePointer& node);
virtual int sendSpecialPacket(OctreeQueryNode* queryNode, const SharedNodePointer& node);
virtual int sendSpecialPacket(const SharedNodePointer& node, OctreeQueryNode* queryNode, int& packetsSent);
private:
bool _sendEnvironments;