//// AvatarMixer.cpp// assignment-client/src/avatars//// Created by Stephen Birarda on 9/5/13.// Copyright 2013 High Fidelity, Inc.// Copyright 2021 Vircadia contributors.//// Distributed under the Apache License, Version 2.0.// See the accompanying file LICENSE or http://www.apache.org/licenses/LICENSE-2.0.html// #include "AvatarMixer.h" #include <cfloat>#include <chrono>#include <memory>#include <random>#include <thread> #include <QtCore/QCoreApplication>#include <QtCore/QDateTime>#include <QtCore/QJsonObject>#include <QtCore/QRegularExpression>#include <QtCore/QTimer>#include <QtCore/QThread>#include <QtCore/QJsonDocument>#include <shared/QtHelpers.h> #include <AABox.h>#include <AvatarLogging.h>#include <LogHandler.h>#include <NodeList.h>#include <udt/PacketHeaders.h>#include <SharedUtil.h>#include <UUID.h>#include <TryLocker.h>#include "../AssignmentDynamicFactory.h"#include "../entities/AssignmentParentFinder.h"#include <model-networking/ModelCache.h>#include <hfm/ModelFormatRegistry.h> const QString AVATAR_MIXER_LOGGING_NAME = "avatar-mixer"; // FIXME - what we'd actually like to do is send to users at ~50% of their present rate down to 30hz. Assume 90 for now.const int AVATAR_MIXER_BROADCAST_FRAMES_PER_SECOND = 45; const QRegularExpression AvatarMixer::suffixedNamePattern { R"(^\s*(.+)\s*_(\d)+\s*$)" }; // Lexicographic comparison:bool AvatarMixer::SessionDisplayName::operator<(const SessionDisplayName& rhs) const { if (_baseName < rhs._baseName) { return true; } else if (rhs._baseName < _baseName) { return false; } else { return _suffix < rhs._suffix; }} AvatarMixer::AvatarMixer(ReceivedMessage& message) : ThreadedAssignment(message), _workerPool(&_workerSharedData){ DependencyManager::registerInheritance<EntityDynamicFactoryInterface, AssignmentDynamicFactory>(); DependencyManager::set<AssignmentDynamicFactory>(); DependencyManager::set<ModelFormatRegistry>(); DependencyManager::set<ModelCache>(); DependencyManager::set<ResourceCacheSharedItems>(); DependencyManager::set<ResourceManager>(); // make sure we hear about node kills so we can tell the other nodes connect(DependencyManager::get<NodeList>().data(), &NodeList::nodeKilled, this, &AvatarMixer::handleAvatarKilled); auto& packetReceiver = DependencyManager::get<NodeList>()->getPacketReceiver(); packetReceiver.registerListener(PacketType::AvatarData, PacketReceiver::makeSourcedListenerReference<AvatarMixer>(this, &AvatarMixer::queueIncomingPacket)); packetReceiver.registerListener(PacketType::AdjustAvatarSorting, PacketReceiver::makeSourcedListenerReference<AvatarMixer>(this, &AvatarMixer::handleAdjustAvatarSorting)); packetReceiver.registerListener(PacketType::AvatarQuery, PacketReceiver::makeSourcedListenerReference<AvatarMixer>(this, &AvatarMixer::handleAvatarQueryPacket)); packetReceiver.registerListener(PacketType::AvatarIdentity, PacketReceiver::makeSourcedListenerReference<AvatarMixer>(this, &AvatarMixer::handleAvatarIdentityPacket)); packetReceiver.registerListener(PacketType::KillAvatar, PacketReceiver::makeSourcedListenerReference<AvatarMixer>(this, &AvatarMixer::handleKillAvatarPacket)); packetReceiver.registerListener(PacketType::NodeIgnoreRequest, PacketReceiver::makeSourcedListenerReference<AvatarMixer>(this, &AvatarMixer::handleNodeIgnoreRequestPacket)); packetReceiver.registerListener(PacketType::RadiusIgnoreRequest, PacketReceiver::makeSourcedListenerReference<AvatarMixer>(this, &AvatarMixer::handleRadiusIgnoreRequestPacket)); packetReceiver.registerListener(PacketType::RequestsDomainListData, PacketReceiver::makeSourcedListenerReference<AvatarMixer>(this, &AvatarMixer::handleRequestsDomainListDataPacket)); packetReceiver.registerListener(PacketType::SetAvatarTraits, PacketReceiver::makeSourcedListenerReference<AvatarMixer>(this, &AvatarMixer::queueIncomingPacket)); packetReceiver.registerListener(PacketType::BulkAvatarTraitsAck, PacketReceiver::makeSourcedListenerReference<AvatarMixer>(this, &AvatarMixer::queueIncomingPacket)); packetReceiver.registerListenerForTypes({ PacketType::OctreeStats, PacketType::EntityData, PacketType::EntityErase }, PacketReceiver::makeSourcedListenerReference<AvatarMixer>(this, &AvatarMixer::handleOctreePacket)); packetReceiver.registerListenerForTypes({ PacketType::ReplicatedAvatarIdentity, PacketType::ReplicatedKillAvatar }, PacketReceiver::makeUnsourcedListenerReference<AvatarMixer>(this, &AvatarMixer::handleReplicatedPacket)); packetReceiver.registerListener(PacketType::ReplicatedBulkAvatarData, PacketReceiver::makeUnsourcedListenerReference<AvatarMixer>(this, &AvatarMixer::handleReplicatedBulkAvatarPacket)); auto nodeList = DependencyManager::get<NodeList>(); connect(nodeList.data(), &NodeList::packetVersionMismatch, this, &AvatarMixer::handlePacketVersionMismatch); connect(nodeList.data(), &NodeList::nodeAdded, this, [this](const SharedNodePointer& node) { if (node->getType() == NodeType::DownstreamAvatarMixer) { getOrCreateClientData(node); } });} SharedNodePointer addOrUpdateReplicatedNode(const QUuid& nodeID, const SockAddr& senderSockAddr) { auto replicatedNode = DependencyManager::get<NodeList>()->addOrUpdateNode(nodeID, NodeType::Agent, senderSockAddr, senderSockAddr, Node::NULL_LOCAL_ID, true, true); replicatedNode->setLastHeardMicrostamp(usecTimestampNow()); return replicatedNode;} void AvatarMixer::handleReplicatedPacket(QSharedPointer<ReceivedMessage> message) { auto nodeList = DependencyManager::get<NodeList>(); auto nodeID = QUuid::fromRfc4122(message->peek(NUM_BYTES_RFC4122_UUID)); SharedNodePointer replicatedNode; if (message->getType() == PacketType::ReplicatedKillAvatar) { // this is a kill packet, which we should only process if we already have the node in our list // since it of course does not make sense to add a node just to remove it an instant later replicatedNode = nodeList->nodeWithUUID(nodeID); if (!replicatedNode) { return; } } else { replicatedNode = addOrUpdateReplicatedNode(nodeID, message->getSenderSockAddr()); } // we better have a node to work with at this point assert(replicatedNode); if (message->getType() == PacketType::ReplicatedAvatarIdentity) { handleAvatarIdentityPacket(message, replicatedNode); } else if (message->getType() == PacketType::ReplicatedKillAvatar) { handleKillAvatarPacket(message, replicatedNode); }} void AvatarMixer::handleReplicatedBulkAvatarPacket(QSharedPointer<ReceivedMessage> message) { while (message->getBytesLeftToRead()) { // first, grab the node ID for this replicated avatar // Node ID is now part of user data, since ReplicatedBulkAvatarPacket is non-sourced. auto nodeID = QUuid::fromRfc4122(message->readWithoutCopy(NUM_BYTES_RFC4122_UUID)); // make sure we have an upstream replicated node that matches auto replicatedNode = addOrUpdateReplicatedNode(nodeID, message->getSenderSockAddr()); // grab the size of the avatar byte array so we know how much to read quint16 avatarByteArraySize; message->readPrimitive(&avatarByteArraySize); // read the avatar byte array auto avatarByteArray = message->read(avatarByteArraySize); // construct a "fake" avatar data received message from the byte array and packet list information auto replicatedMessage = QSharedPointer<ReceivedMessage>::create(avatarByteArray, PacketType::AvatarData, versionForPacketType(PacketType::AvatarData), message->getSenderSockAddr(), Node::NULL_LOCAL_ID); // queue up the replicated avatar data with the client data for the replicated node auto start = usecTimestampNow(); getOrCreateClientData(replicatedNode)->queuePacket(replicatedMessage, replicatedNode); auto end = usecTimestampNow(); _queueIncomingPacketElapsedTime += (end - start); }} void AvatarMixer::optionallyReplicatePacket(ReceivedMessage& message, const Node& node) { // first, make sure that this is a packet from a node we are supposed to replicate if (node.isReplicated()) { // check if this is a packet type we replicate // which means it must be a packet type present in REPLICATED_PACKET_MAPPING or must be the // replicated version of one of those packet types PacketType replicatedType = PacketTypeEnum::getReplicatedPacketMapping().value(message.getType()); if (replicatedType == PacketType::Unknown) { if (PacketTypeEnum::getReplicatedPacketMapping().key(message.getType()) != PacketType::Unknown) { replicatedType = message.getType(); } else { qDebug() << __FUNCTION__ << "called without replicatable packet type - returning"; return; } } std::unique_ptr<NLPacket> packet; auto nodeList = DependencyManager::get<NodeList>(); nodeList->eachMatchingNode([&](const SharedNodePointer& downstreamNode) { return shouldReplicateTo(node, *downstreamNode); }, [&](const SharedNodePointer& node) { if (!packet) { // construct an NLPacket to send to the replicant that has the contents of the received packet packet = NLPacket::create(replicatedType, message.getSize()); packet->write(message.getMessage()); } nodeList->sendUnreliablePacket(*packet, *node); }); }} void AvatarMixer::queueIncomingPacket(QSharedPointer<ReceivedMessage> message, SharedNodePointer node) { auto start = usecTimestampNow(); getOrCreateClientData(node)->queuePacket(message, node); auto end = usecTimestampNow(); _queueIncomingPacketElapsedTime += (end - start);} void AvatarMixer::sendIdentityPacket(AvatarMixerClientData* nodeData, const SharedNodePointer& destinationNode) { if (destinationNode->getType() == NodeType::Agent && !destinationNode->isUpstream()) { QByteArray individualData = nodeData->getAvatar().identityByteArray(); individualData.replace(0, NUM_BYTES_RFC4122_UUID, nodeData->getNodeID().toRfc4122()); auto identityPackets = NLPacketList::create(PacketType::AvatarIdentity, QByteArray(), true, true); identityPackets->write(individualData); DependencyManager::get<NodeList>()->sendPacketList(std::move(identityPackets), *destinationNode); ++_sumIdentityPackets; }} std::chrono::microseconds AvatarMixer::timeFrame(p_high_resolution_clock::time_point& timestamp) { // advance the next frame auto nextTimestamp = timestamp + std::chrono::microseconds((int)((float)USECS_PER_SECOND / (float)AVATAR_MIXER_BROADCAST_FRAMES_PER_SECOND)); auto now = p_high_resolution_clock::now(); // compute how long the last frame took auto duration = std::chrono::duration_cast<std::chrono::microseconds>(now - timestamp); // set the new frame timestamp timestamp = std::max(now, nextTimestamp); // sleep until the next frame should start // WIN32 sleep_until is broken until VS2015 Update 2 // instead, std::max (above) guarantees that timestamp >= now, so we can sleep_for std::this_thread::sleep_for(timestamp - now); return duration;} void AvatarMixer::start() { auto nodeList = DependencyManager::get<NodeList>(); unsigned int frame = 1; auto frameTimestamp = p_high_resolution_clock::now(); while (!_isFinished) { auto frameDuration = timeFrame(frameTimestamp); // calculates last frame duration and sleeps remainder of target amount throttle(frameDuration, frame); // determines _throttlingRatio for upcoming mix frame int lockWait, nodeTransform, functor; // Set our query each frame { _entityViewer.queryOctree(); } // Dirty the hero status if there's been an entity change. { if (_dirtyHeroStatus) { _dirtyHeroStatus = false; nodeList->nestedEach([](NodeList::const_iterator cbegin, NodeList::const_iterator cend) { std::for_each(cbegin, cend, [](const SharedNodePointer& node) { if (node->getType() == NodeType::Agent) { NodeData* nodeData = node->getLinkedData(); if (nodeData) { auto& avatar = static_cast<AvatarMixerClientData*>(nodeData)->getAvatar(); avatar.setNeedsHeroCheck(); } } }); }); } } // Allow nodes to process any pending/queued packets across our worker threads { auto start = usecTimestampNow(); nodeList->nestedEach([&](NodeList::const_iterator cbegin, NodeList::const_iterator cend) { auto end = usecTimestampNow(); _processQueuedAvatarDataPacketsLockWaitElapsedTime += (end - start); _workerPool.processIncomingPackets(cbegin, cend); }, &lockWait, &nodeTransform, &functor); auto end = usecTimestampNow(); _processQueuedAvatarDataPacketsElapsedTime += (end - start); _broadcastAvatarDataLockWait += lockWait; _broadcastAvatarDataNodeTransform += nodeTransform; _broadcastAvatarDataNodeFunctor += functor; } // process pending display names... this doesn't currently run on multiple threads, because it // side-effects the mixer's data, which is fine because it's a very low cost operation { auto start = usecTimestampNow(); nodeList->nestedEach([&](NodeList::const_iterator cbegin, NodeList::const_iterator cend) { std::for_each(cbegin, cend, [&](const SharedNodePointer& node) { if (node->getType() == NodeType::Agent) { manageIdentityData(node); } ++_sumListeners; }); }, &lockWait, &nodeTransform, &functor); auto end = usecTimestampNow(); _displayNameManagementElapsedTime += (end - start); _broadcastAvatarDataLockWait += lockWait; _broadcastAvatarDataNodeTransform += nodeTransform; _broadcastAvatarDataNodeFunctor += functor; } // this is where we need to put the real work... { auto start = usecTimestampNow(); nodeList->nestedEach([&](NodeList::const_iterator cbegin, NodeList::const_iterator cend) { auto start = usecTimestampNow(); _workerPool.broadcastAvatarData(cbegin, cend, _lastFrameTimestamp, _maxKbpsPerNode, _throttlingRatio); auto end = usecTimestampNow(); _broadcastAvatarDataInner += (end - start); }, &lockWait, &nodeTransform, &functor); auto end = usecTimestampNow(); _broadcastAvatarDataElapsedTime += (end - start); _broadcastAvatarDataLockWait += lockWait; _broadcastAvatarDataNodeTransform += nodeTransform; _broadcastAvatarDataNodeFunctor += functor; } ++frame; ++_numTightLoopFrames; _loopRate.increment(); // play nice with qt event-looping { // since we're a while loop we need to yield to qt's event processing auto start = usecTimestampNow(); QCoreApplication::processEvents(); if (_isFinished) { // alert qt eventing that this is finished QCoreApplication::sendPostedEvents(this, QEvent::DeferredDelete); break; } auto end = usecTimestampNow(); _processEventsElapsedTime += (end - start); } _lastFrameTimestamp = frameTimestamp; }} // NOTE: nodeData->getAvatar() might be side effected, must be called when access to node/nodeData// is guaranteed to not be accessed by other threadvoid AvatarMixer::manageIdentityData(const SharedNodePointer& node) { AvatarMixerClientData* nodeData = reinterpret_cast<AvatarMixerClientData*>(node->getLinkedData()); // there is no need to manage identity data we haven't received yet // so bail early if we've never received an identity packet for this avatar if (!nodeData || !nodeData->getAvatar().hasProcessedFirstIdentity() || !node->getActiveSocket()) { return; } MixerAvatar& avatar = nodeData->getAvatar(); bool sendIdentity = avatar.needsIdentityUpdate(); if (sendIdentity) { nodeData->flagIdentityChange(); } if (nodeData->getAvatarSessionDisplayNameMustChange()) { const QString& existingBaseDisplayName = avatar.getSessionDisplayName(); if (!existingBaseDisplayName.isEmpty()) { SessionDisplayName existingDisplayName { existingBaseDisplayName }; auto suffixMatch = suffixedNamePattern.match(existingBaseDisplayName); if (suffixMatch.hasMatch()) { existingDisplayName._baseName = suffixMatch.captured(1); existingDisplayName._suffix = suffixMatch.captured(2).toInt(); } _sessionDisplayNames.erase(existingDisplayName); } QString baseName = avatar.getDisplayName().trimmed(); const QRegularExpression curses { "fuck|shit|damn|cock|cunt" }; // POC. We may eventually want something much more elaborate (subscription?). baseName = baseName.replace(curses, "*"); // Replace rather than remove, so that people have a clue that the person's a jerk. static const QRegularExpression trailingDigits { R"(\s*(_\d+\s*)?(\s*\n[^$]*)?$)" }; // trailing whitespace "_123" and any subsequent lines baseName = baseName.remove(trailingDigits); if (baseName.isEmpty()) { baseName = "anonymous"; } SessionDisplayName newDisplayName { baseName }; auto nameIter = _sessionDisplayNames.lower_bound(newDisplayName); if (nameIter != _sessionDisplayNames.end() && nameIter->_baseName == baseName) { // Existing instance(s) of name; find first free suffix while (*nameIter == newDisplayName && ++newDisplayName._suffix && ++nameIter != _sessionDisplayNames.end()) ; } _sessionDisplayNames.insert(newDisplayName); QString sessionDisplayName = (newDisplayName._suffix > 0) ? baseName + "_" + QString::number(newDisplayName._suffix) : baseName; avatar.setSessionDisplayName(sessionDisplayName); nodeData->setBaseDisplayName(baseName); nodeData->flagIdentityChange(); nodeData->setAvatarSessionDisplayNameMustChange(false); sendIdentity = true; qCDebug(avatars) << "Giving session display name" << sessionDisplayName << "to node with ID" << node->getUUID(); } if (sendIdentity && !node->isUpstream()) { sendIdentityPacket(nodeData, node); // Tell node whose name changed about its new session display name or avatar. // since this packet includes a change to either the skeleton model URL or the display name // it needs a new sequence number avatar.pushIdentitySequenceNumber(); // tell node whose name changed about its new session display name or avatar. sendIdentityPacket(nodeData, node); avatar.setNeedsIdentityUpdate(false); }} void AvatarMixer::throttle(std::chrono::microseconds duration, int frame) { // throttle using a modified proportional-integral controller const float FRAME_TIME = USECS_PER_SECOND / AVATAR_MIXER_BROADCAST_FRAMES_PER_SECOND; float mixRatio = duration.count() / FRAME_TIME; // constants are determined based on a "regular" 16-CPU EC2 server // target different mix and backoff ratios (they also have different backoff rates) // this is to prevent oscillation, and encourage throttling to find a steady state const float TARGET = 0.9f; // on a "regular" machine with 100 avatars, this is the largest value where // - overthrottling can be recovered // - oscillations will not occur after the recovery const float BACKOFF_TARGET = 0.44f; // the mixer is known to struggle at about 150 on a "regular" machine // so throttle 2/150 the streams to ensure smooth mixing (throttling is linear) const float STRUGGLES_AT = 150.0f; const float THROTTLE_RATE = 2 / STRUGGLES_AT; const float BACKOFF_RATE = THROTTLE_RATE / 4; // recovery should be bounded so that large changes in user count is a tolerable experience // throttling is linear, so most cases will not need a full recovery const int RECOVERY_TIME = 180; // weight more recent frames to determine if throttling is necessary, const int TRAILING_FRAMES = (int)(100 * RECOVERY_TIME * BACKOFF_RATE); const float CURRENT_FRAME_RATIO = 1.0f / TRAILING_FRAMES; const float PREVIOUS_FRAMES_RATIO = 1.0f - CURRENT_FRAME_RATIO; _trailingMixRatio = PREVIOUS_FRAMES_RATIO * _trailingMixRatio + CURRENT_FRAME_RATIO * mixRatio; if (frame % TRAILING_FRAMES == 0) { if (_trailingMixRatio > TARGET) { int proportionalTerm = 1 + (_trailingMixRatio - TARGET) / 0.1f; _throttlingRatio += THROTTLE_RATE * proportionalTerm; _throttlingRatio = std::min(_throttlingRatio, 1.0f); qDebug("avatar-mixer is struggling (%f mix/sleep) - throttling %f of streams", (double)_trailingMixRatio, (double)_throttlingRatio); } else if (_throttlingRatio > 0.0f && _trailingMixRatio <= BACKOFF_TARGET) { int proportionalTerm = 1 + (TARGET - _trailingMixRatio) / 0.2f; _throttlingRatio -= BACKOFF_RATE * proportionalTerm; _throttlingRatio = std::max(_throttlingRatio, 0.0f); qDebug("avatar-mixer is recovering (%f mix/sleep) - throttling %f of streams", (double)_trailingMixRatio, (double)_throttlingRatio); } }} void AvatarMixer::handleAvatarKilled(SharedNodePointer avatarNode) { if (avatarNode->getType() == NodeType::Agent && avatarNode->getLinkedData()) { auto nodeList = DependencyManager::get<NodeList>(); { // decrement sessionDisplayNames table and possibly remove QMutexLocker nodeDataLocker(&avatarNode->getLinkedData()->getMutex()); AvatarMixerClientData* nodeData = dynamic_cast<AvatarMixerClientData*>(avatarNode->getLinkedData()); const QString& displayName = nodeData->getAvatar().getSessionDisplayName(); SessionDisplayName exitingDisplayName { displayName };Lines 1–500 of 1164. Load more
// // AvatarMixer.cpp // assignment-client/src/avatars // // Created by Stephen Birarda on 9/5/13. // Copyright 2013 High Fidelity, Inc. // Copyright 2021 Vircadia contributors. // // Distributed under the Apache License, Version 2.0. // See the accompanying file LICENSE or http://www.apache.org/licenses/LICENSE-2.0.html // #include "AvatarMixer.h" #include <cfloat> #include <chrono> #include <memory> #include <random> #include <thread> #include <QtCore/QCoreApplication> #include <QtCore/QDateTime> #include <QtCore/QJsonObject> #include <QtCore/QRegularExpression> #include <QtCore/QTimer> #include <QtCore/QThread> #include <QtCore/QJsonDocument> #include <shared/QtHelpers.h> #include <AABox.h> #include <AvatarLogging.h> #include <LogHandler.h> #include <NodeList.h> #include <udt/PacketHeaders.h> #include <SharedUtil.h> #include <UUID.h> #include <TryLocker.h> #include "../AssignmentDynamicFactory.h" #include "../entities/AssignmentParentFinder.h" #include <model-networking/ModelCache.h> #include <hfm/ModelFormatRegistry.h> const QString AVATAR_MIXER_LOGGING_NAME = "avatar-mixer"; // FIXME - what we'd actually like to do is send to users at ~50% of their present rate down to 30hz. Assume 90 for now. const int AVATAR_MIXER_BROADCAST_FRAMES_PER_SECOND = 45; const QRegularExpression AvatarMixer::suffixedNamePattern { R"(^\s*(.+)\s*_(\d)+\s*$)" }; // Lexicographic comparison: bool AvatarMixer::SessionDisplayName::operator<(const SessionDisplayName& rhs) const { if (_baseName < rhs._baseName) { return true; } else if (rhs._baseName < _baseName) { return false; } else { return _suffix < rhs._suffix; } } AvatarMixer::AvatarMixer(ReceivedMessage& message) : ThreadedAssignment(message), _workerPool(&_workerSharedData) { DependencyManager::registerInheritance<EntityDynamicFactoryInterface, AssignmentDynamicFactory>(); DependencyManager::set<AssignmentDynamicFactory>(); DependencyManager::set<ModelFormatRegistry>(); DependencyManager::set<ModelCache>(); DependencyManager::set<ResourceCacheSharedItems>(); DependencyManager::set<ResourceManager>(); // make sure we hear about node kills so we can tell the other nodes connect(DependencyManager::get<NodeList>().data(), &NodeList::nodeKilled, this, &AvatarMixer::handleAvatarKilled); auto& packetReceiver = DependencyManager::get<NodeList>()->getPacketReceiver(); packetReceiver.registerListener(PacketType::AvatarData, PacketReceiver::makeSourcedListenerReference<AvatarMixer>(this, &AvatarMixer::queueIncomingPacket)); packetReceiver.registerListener(PacketType::AdjustAvatarSorting, PacketReceiver::makeSourcedListenerReference<AvatarMixer>(this, &AvatarMixer::handleAdjustAvatarSorting)); packetReceiver.registerListener(PacketType::AvatarQuery, PacketReceiver::makeSourcedListenerReference<AvatarMixer>(this, &AvatarMixer::handleAvatarQueryPacket)); packetReceiver.registerListener(PacketType::AvatarIdentity, PacketReceiver::makeSourcedListenerReference<AvatarMixer>(this, &AvatarMixer::handleAvatarIdentityPacket)); packetReceiver.registerListener(PacketType::KillAvatar, PacketReceiver::makeSourcedListenerReference<AvatarMixer>(this, &AvatarMixer::handleKillAvatarPacket)); packetReceiver.registerListener(PacketType::NodeIgnoreRequest, PacketReceiver::makeSourcedListenerReference<AvatarMixer>(this, &AvatarMixer::handleNodeIgnoreRequestPacket)); packetReceiver.registerListener(PacketType::RadiusIgnoreRequest, PacketReceiver::makeSourcedListenerReference<AvatarMixer>(this, &AvatarMixer::handleRadiusIgnoreRequestPacket)); packetReceiver.registerListener(PacketType::RequestsDomainListData, PacketReceiver::makeSourcedListenerReference<AvatarMixer>(this, &AvatarMixer::handleRequestsDomainListDataPacket)); packetReceiver.registerListener(PacketType::SetAvatarTraits, PacketReceiver::makeSourcedListenerReference<AvatarMixer>(this, &AvatarMixer::queueIncomingPacket)); packetReceiver.registerListener(PacketType::BulkAvatarTraitsAck, PacketReceiver::makeSourcedListenerReference<AvatarMixer>(this, &AvatarMixer::queueIncomingPacket)); packetReceiver.registerListenerForTypes({ PacketType::OctreeStats, PacketType::EntityData, PacketType::EntityErase }, PacketReceiver::makeSourcedListenerReference<AvatarMixer>(this, &AvatarMixer::handleOctreePacket)); packetReceiver.registerListenerForTypes({ PacketType::ReplicatedAvatarIdentity, PacketType::ReplicatedKillAvatar }, PacketReceiver::makeUnsourcedListenerReference<AvatarMixer>(this, &AvatarMixer::handleReplicatedPacket)); packetReceiver.registerListener(PacketType::ReplicatedBulkAvatarData, PacketReceiver::makeUnsourcedListenerReference<AvatarMixer>(this, &AvatarMixer::handleReplicatedBulkAvatarPacket)); auto nodeList = DependencyManager::get<NodeList>(); connect(nodeList.data(), &NodeList::packetVersionMismatch, this, &AvatarMixer::handlePacketVersionMismatch); connect(nodeList.data(), &NodeList::nodeAdded, this, [this](const SharedNodePointer& node) { if (node->getType() == NodeType::DownstreamAvatarMixer) { getOrCreateClientData(node); } }); } SharedNodePointer addOrUpdateReplicatedNode(const QUuid& nodeID, const SockAddr& senderSockAddr) { auto replicatedNode = DependencyManager::get<NodeList>()->addOrUpdateNode(nodeID, NodeType::Agent, senderSockAddr, senderSockAddr, Node::NULL_LOCAL_ID, true, true); replicatedNode->setLastHeardMicrostamp(usecTimestampNow()); return replicatedNode; } void AvatarMixer::handleReplicatedPacket(QSharedPointer<ReceivedMessage> message) { auto nodeList = DependencyManager::get<NodeList>(); auto nodeID = QUuid::fromRfc4122(message->peek(NUM_BYTES_RFC4122_UUID)); SharedNodePointer replicatedNode; if (message->getType() == PacketType::ReplicatedKillAvatar) { // this is a kill packet, which we should only process if we already have the node in our list // since it of course does not make sense to add a node just to remove it an instant later replicatedNode = nodeList->nodeWithUUID(nodeID); if (!replicatedNode) { return; } } else { replicatedNode = addOrUpdateReplicatedNode(nodeID, message->getSenderSockAddr()); } // we better have a node to work with at this point assert(replicatedNode); if (message->getType() == PacketType::ReplicatedAvatarIdentity) { handleAvatarIdentityPacket(message, replicatedNode); } else if (message->getType() == PacketType::ReplicatedKillAvatar) { handleKillAvatarPacket(message, replicatedNode); } } void AvatarMixer::handleReplicatedBulkAvatarPacket(QSharedPointer<ReceivedMessage> message) { while (message->getBytesLeftToRead()) { // first, grab the node ID for this replicated avatar // Node ID is now part of user data, since ReplicatedBulkAvatarPacket is non-sourced. auto nodeID = QUuid::fromRfc4122(message->readWithoutCopy(NUM_BYTES_RFC4122_UUID)); // make sure we have an upstream replicated node that matches auto replicatedNode = addOrUpdateReplicatedNode(nodeID, message->getSenderSockAddr()); // grab the size of the avatar byte array so we know how much to read quint16 avatarByteArraySize; message->readPrimitive(&avatarByteArraySize); // read the avatar byte array auto avatarByteArray = message->read(avatarByteArraySize); // construct a "fake" avatar data received message from the byte array and packet list information auto replicatedMessage = QSharedPointer<ReceivedMessage>::create(avatarByteArray, PacketType::AvatarData, versionForPacketType(PacketType::AvatarData), message->getSenderSockAddr(), Node::NULL_LOCAL_ID); // queue up the replicated avatar data with the client data for the replicated node auto start = usecTimestampNow(); getOrCreateClientData(replicatedNode)->queuePacket(replicatedMessage, replicatedNode); auto end = usecTimestampNow(); _queueIncomingPacketElapsedTime += (end - start); } } void AvatarMixer::optionallyReplicatePacket(ReceivedMessage& message, const Node& node) { // first, make sure that this is a packet from a node we are supposed to replicate if (node.isReplicated()) { // check if this is a packet type we replicate // which means it must be a packet type present in REPLICATED_PACKET_MAPPING or must be the // replicated version of one of those packet types PacketType replicatedType = PacketTypeEnum::getReplicatedPacketMapping().value(message.getType()); if (replicatedType == PacketType::Unknown) { if (PacketTypeEnum::getReplicatedPacketMapping().key(message.getType()) != PacketType::Unknown) { replicatedType = message.getType(); } else { qDebug() << __FUNCTION__ << "called without replicatable packet type - returning"; return; } } std::unique_ptr<NLPacket> packet; auto nodeList = DependencyManager::get<NodeList>(); nodeList->eachMatchingNode([&](const SharedNodePointer& downstreamNode) { return shouldReplicateTo(node, *downstreamNode); }, [&](const SharedNodePointer& node) { if (!packet) { // construct an NLPacket to send to the replicant that has the contents of the received packet packet = NLPacket::create(replicatedType, message.getSize()); packet->write(message.getMessage()); } nodeList->sendUnreliablePacket(*packet, *node); }); } } void AvatarMixer::queueIncomingPacket(QSharedPointer<ReceivedMessage> message, SharedNodePointer node) { auto start = usecTimestampNow(); getOrCreateClientData(node)->queuePacket(message, node); auto end = usecTimestampNow(); _queueIncomingPacketElapsedTime += (end - start); } void AvatarMixer::sendIdentityPacket(AvatarMixerClientData* nodeData, const SharedNodePointer& destinationNode) { if (destinationNode->getType() == NodeType::Agent && !destinationNode->isUpstream()) { QByteArray individualData = nodeData->getAvatar().identityByteArray(); individualData.replace(0, NUM_BYTES_RFC4122_UUID, nodeData->getNodeID().toRfc4122()); auto identityPackets = NLPacketList::create(PacketType::AvatarIdentity, QByteArray(), true, true); identityPackets->write(individualData); DependencyManager::get<NodeList>()->sendPacketList(std::move(identityPackets), *destinationNode); ++_sumIdentityPackets; } } std::chrono::microseconds AvatarMixer::timeFrame(p_high_resolution_clock::time_point& timestamp) { // advance the next frame auto nextTimestamp = timestamp + std::chrono::microseconds((int)((float)USECS_PER_SECOND / (float)AVATAR_MIXER_BROADCAST_FRAMES_PER_SECOND)); auto now = p_high_resolution_clock::now(); // compute how long the last frame took auto duration = std::chrono::duration_cast<std::chrono::microseconds>(now - timestamp); // set the new frame timestamp timestamp = std::max(now, nextTimestamp); // sleep until the next frame should start // WIN32 sleep_until is broken until VS2015 Update 2 // instead, std::max (above) guarantees that timestamp >= now, so we can sleep_for std::this_thread::sleep_for(timestamp - now); return duration; } void AvatarMixer::start() { auto nodeList = DependencyManager::get<NodeList>(); unsigned int frame = 1; auto frameTimestamp = p_high_resolution_clock::now(); while (!_isFinished) { auto frameDuration = timeFrame(frameTimestamp); // calculates last frame duration and sleeps remainder of target amount throttle(frameDuration, frame); // determines _throttlingRatio for upcoming mix frame int lockWait, nodeTransform, functor; // Set our query each frame { _entityViewer.queryOctree(); } // Dirty the hero status if there's been an entity change. { if (_dirtyHeroStatus) { _dirtyHeroStatus = false; nodeList->nestedEach([](NodeList::const_iterator cbegin, NodeList::const_iterator cend) { std::for_each(cbegin, cend, [](const SharedNodePointer& node) { if (node->getType() == NodeType::Agent) { NodeData* nodeData = node->getLinkedData(); if (nodeData) { auto& avatar = static_cast<AvatarMixerClientData*>(nodeData)->getAvatar(); avatar.setNeedsHeroCheck(); } } }); }); } } // Allow nodes to process any pending/queued packets across our worker threads { auto start = usecTimestampNow(); nodeList->nestedEach([&](NodeList::const_iterator cbegin, NodeList::const_iterator cend) { auto end = usecTimestampNow(); _processQueuedAvatarDataPacketsLockWaitElapsedTime += (end - start); _workerPool.processIncomingPackets(cbegin, cend); }, &lockWait, &nodeTransform, &functor); auto end = usecTimestampNow(); _processQueuedAvatarDataPacketsElapsedTime += (end - start); _broadcastAvatarDataLockWait += lockWait; _broadcastAvatarDataNodeTransform += nodeTransform; _broadcastAvatarDataNodeFunctor += functor; } // process pending display names... this doesn't currently run on multiple threads, because it // side-effects the mixer's data, which is fine because it's a very low cost operation { auto start = usecTimestampNow(); nodeList->nestedEach([&](NodeList::const_iterator cbegin, NodeList::const_iterator cend) { std::for_each(cbegin, cend, [&](const SharedNodePointer& node) { if (node->getType() == NodeType::Agent) { manageIdentityData(node); } ++_sumListeners; }); }, &lockWait, &nodeTransform, &functor); auto end = usecTimestampNow(); _displayNameManagementElapsedTime += (end - start); _broadcastAvatarDataLockWait += lockWait; _broadcastAvatarDataNodeTransform += nodeTransform; _broadcastAvatarDataNodeFunctor += functor; } // this is where we need to put the real work... { auto start = usecTimestampNow(); nodeList->nestedEach([&](NodeList::const_iterator cbegin, NodeList::const_iterator cend) { auto start = usecTimestampNow(); _workerPool.broadcastAvatarData(cbegin, cend, _lastFrameTimestamp, _maxKbpsPerNode, _throttlingRatio); auto end = usecTimestampNow(); _broadcastAvatarDataInner += (end - start); }, &lockWait, &nodeTransform, &functor); auto end = usecTimestampNow(); _broadcastAvatarDataElapsedTime += (end - start); _broadcastAvatarDataLockWait += lockWait; _broadcastAvatarDataNodeTransform += nodeTransform; _broadcastAvatarDataNodeFunctor += functor; } ++frame; ++_numTightLoopFrames; _loopRate.increment(); // play nice with qt event-looping { // since we're a while loop we need to yield to qt's event processing auto start = usecTimestampNow(); QCoreApplication::processEvents(); if (_isFinished) { // alert qt eventing that this is finished QCoreApplication::sendPostedEvents(this, QEvent::DeferredDelete); break; } auto end = usecTimestampNow(); _processEventsElapsedTime += (end - start); } _lastFrameTimestamp = frameTimestamp; } } // NOTE: nodeData->getAvatar() might be side effected, must be called when access to node/nodeData // is guaranteed to not be accessed by other thread void AvatarMixer::manageIdentityData(const SharedNodePointer& node) { AvatarMixerClientData* nodeData = reinterpret_cast<AvatarMixerClientData*>(node->getLinkedData()); // there is no need to manage identity data we haven't received yet // so bail early if we've never received an identity packet for this avatar if (!nodeData || !nodeData->getAvatar().hasProcessedFirstIdentity() || !node->getActiveSocket()) { return; } MixerAvatar& avatar = nodeData->getAvatar(); bool sendIdentity = avatar.needsIdentityUpdate(); if (sendIdentity) { nodeData->flagIdentityChange(); } if (nodeData->getAvatarSessionDisplayNameMustChange()) { const QString& existingBaseDisplayName = avatar.getSessionDisplayName(); if (!existingBaseDisplayName.isEmpty()) { SessionDisplayName existingDisplayName { existingBaseDisplayName }; auto suffixMatch = suffixedNamePattern.match(existingBaseDisplayName); if (suffixMatch.hasMatch()) { existingDisplayName._baseName = suffixMatch.captured(1); existingDisplayName._suffix = suffixMatch.captured(2).toInt(); } _sessionDisplayNames.erase(existingDisplayName); } QString baseName = avatar.getDisplayName().trimmed(); const QRegularExpression curses { "fuck|shit|damn|cock|cunt" }; // POC. We may eventually want something much more elaborate (subscription?). baseName = baseName.replace(curses, "*"); // Replace rather than remove, so that people have a clue that the person's a jerk. static const QRegularExpression trailingDigits { R"(\s*(_\d+\s*)?(\s*\n[^$]*)?$)" }; // trailing whitespace "_123" and any subsequent lines baseName = baseName.remove(trailingDigits); if (baseName.isEmpty()) { baseName = "anonymous"; } SessionDisplayName newDisplayName { baseName }; auto nameIter = _sessionDisplayNames.lower_bound(newDisplayName); if (nameIter != _sessionDisplayNames.end() && nameIter->_baseName == baseName) { // Existing instance(s) of name; find first free suffix while (*nameIter == newDisplayName && ++newDisplayName._suffix && ++nameIter != _sessionDisplayNames.end()) ; } _sessionDisplayNames.insert(newDisplayName); QString sessionDisplayName = (newDisplayName._suffix > 0) ? baseName + "_" + QString::number(newDisplayName._suffix) : baseName; avatar.setSessionDisplayName(sessionDisplayName); nodeData->setBaseDisplayName(baseName); nodeData->flagIdentityChange(); nodeData->setAvatarSessionDisplayNameMustChange(false); sendIdentity = true; qCDebug(avatars) << "Giving session display name" << sessionDisplayName << "to node with ID" << node->getUUID(); } if (sendIdentity && !node->isUpstream()) { sendIdentityPacket(nodeData, node); // Tell node whose name changed about its new session display name or avatar. // since this packet includes a change to either the skeleton model URL or the display name // it needs a new sequence number avatar.pushIdentitySequenceNumber(); // tell node whose name changed about its new session display name or avatar. sendIdentityPacket(nodeData, node); avatar.setNeedsIdentityUpdate(false); } } void AvatarMixer::throttle(std::chrono::microseconds duration, int frame) { // throttle using a modified proportional-integral controller const float FRAME_TIME = USECS_PER_SECOND / AVATAR_MIXER_BROADCAST_FRAMES_PER_SECOND; float mixRatio = duration.count() / FRAME_TIME; // constants are determined based on a "regular" 16-CPU EC2 server // target different mix and backoff ratios (they also have different backoff rates) // this is to prevent oscillation, and encourage throttling to find a steady state const float TARGET = 0.9f; // on a "regular" machine with 100 avatars, this is the largest value where // - overthrottling can be recovered // - oscillations will not occur after the recovery const float BACKOFF_TARGET = 0.44f; // the mixer is known to struggle at about 150 on a "regular" machine // so throttle 2/150 the streams to ensure smooth mixing (throttling is linear) const float STRUGGLES_AT = 150.0f; const float THROTTLE_RATE = 2 / STRUGGLES_AT; const float BACKOFF_RATE = THROTTLE_RATE / 4; // recovery should be bounded so that large changes in user count is a tolerable experience // throttling is linear, so most cases will not need a full recovery const int RECOVERY_TIME = 180; // weight more recent frames to determine if throttling is necessary, const int TRAILING_FRAMES = (int)(100 * RECOVERY_TIME * BACKOFF_RATE); const float CURRENT_FRAME_RATIO = 1.0f / TRAILING_FRAMES; const float PREVIOUS_FRAMES_RATIO = 1.0f - CURRENT_FRAME_RATIO; _trailingMixRatio = PREVIOUS_FRAMES_RATIO * _trailingMixRatio + CURRENT_FRAME_RATIO * mixRatio; if (frame % TRAILING_FRAMES == 0) { if (_trailingMixRatio > TARGET) { int proportionalTerm = 1 + (_trailingMixRatio - TARGET) / 0.1f; _throttlingRatio += THROTTLE_RATE * proportionalTerm; _throttlingRatio = std::min(_throttlingRatio, 1.0f); qDebug("avatar-mixer is struggling (%f mix/sleep) - throttling %f of streams", (double)_trailingMixRatio, (double)_throttlingRatio); } else if (_throttlingRatio > 0.0f && _trailingMixRatio <= BACKOFF_TARGET) { int proportionalTerm = 1 + (TARGET - _trailingMixRatio) / 0.2f; _throttlingRatio -= BACKOFF_RATE * proportionalTerm; _throttlingRatio = std::max(_throttlingRatio, 0.0f); qDebug("avatar-mixer is recovering (%f mix/sleep) - throttling %f of streams", (double)_trailingMixRatio, (double)_throttlingRatio); } } } void AvatarMixer::handleAvatarKilled(SharedNodePointer avatarNode) { if (avatarNode->getType() == NodeType::Agent && avatarNode->getLinkedData()) { auto nodeList = DependencyManager::get<NodeList>(); { // decrement sessionDisplayNames table and possibly remove QMutexLocker nodeDataLocker(&avatarNode->getLinkedData()->getMutex()); AvatarMixerClientData* nodeData = dynamic_cast<AvatarMixerClientData*>(avatarNode->getLinkedData()); const QString& displayName = nodeData->getAvatar().getSessionDisplayName(); SessionDisplayName exitingDisplayName { displayName };