Merge pull request #8843 from sethalves/fix-audio-mixer-data-race-0

Fix some audio mixer data races

assignment-client/src/AssignmentClient.cpp

@@ -53,10 +53,13 @@ AssignmentClient::AssignmentClient(Assignment::Type requestAssignmentType, QStri
     QSettings::setDefaultFormat(QSettings::IniFormat);
 
     DependencyManager::set<AccountManager>();
- 
+
     auto scriptableAvatar = DependencyManager::set<ScriptableAvatar>();
     auto addressManager = DependencyManager::set<AddressManager>();
-    auto scriptEngines = DependencyManager::set<ScriptEngines>();
+
+    if (requestAssignmentType == Assignment::AgentType) {
+        auto scriptEngines = DependencyManager::set<ScriptEngines>();
+    }
 
     // create a NodeList as an unassigned client, must be after addressManager
     auto nodeList = DependencyManager::set<NodeList>(NodeType::Unassigned, listenPort);

assignment-client/src/audio/AudioMixer.cpp

@@ -636,7 +636,7 @@ QString AudioMixer::percentageForMixStats(int counter) {
 }
 
 void AudioMixer::sendStatsPacket() {
-    static QJsonObject statsObject;
+    QJsonObject statsObject;
 
     statsObject["useDynamicJitterBuffers"] = _numStaticJitterFrames == -1;
     statsObject["trailing_sleep_percentage"] = _trailingSleepRatio * 100.0f;

libraries/networking/src/LimitedNodeList.h

@@ -321,9 +321,8 @@ protected:
 
     PacketReceiver* _packetReceiver;
 
-    // XXX can BandwidthRecorder be used for this?
-    int _numCollectedPackets;
-    int _numCollectedBytes;
+    std::atomic<int> _numCollectedPackets;
+    std::atomic<int> _numCollectedBytes;
 
     QElapsedTimer _packetStatTimer;
     NodePermissions _permissions;

libraries/networking/src/NodeList.cpp

@@ -127,19 +127,30 @@ NodeList::NodeList(char newOwnerType, int socketListenPort, int dtlsListenPort)
     packetReceiver.registerListener(PacketType::DomainServerRemovedNode, this, "processDomainServerRemovedNode");
 }
 
-qint64 NodeList::sendStats(const QJsonObject& statsObject, const HifiSockAddr& destination) {
+qint64 NodeList::sendStats(QJsonObject statsObject, HifiSockAddr destination) {
+    if (thread() != QThread::currentThread()) {
+        QMetaObject::invokeMethod(this, "sendStats", Qt::QueuedConnection,
+                                  Q_ARG(QJsonObject, statsObject),
+                                  Q_ARG(HifiSockAddr, destination));
+        return 0;
+    }
+
     auto statsPacketList = NLPacketList::create(PacketType::NodeJsonStats, QByteArray(), true, true);
 
     QJsonDocument jsonDocument(statsObject);
     statsPacketList->write(jsonDocument.toBinaryData());
 
     sendPacketList(std::move(statsPacketList), destination);
-
-    // enumerate the resulting strings, breaking them into MTU sized packets
     return 0;
 }
 
-qint64 NodeList::sendStatsToDomainServer(const QJsonObject& statsObject) {
+qint64 NodeList::sendStatsToDomainServer(QJsonObject statsObject) {
+    if (thread() != QThread::currentThread()) {
+        QMetaObject::invokeMethod(this, "sendStatsToDomainServer", Qt::QueuedConnection,
+                                  Q_ARG(QJsonObject, statsObject));
+        return 0;
+    }
+
     return sendStats(statsObject, _domainHandler.getSockAddr());
 }
 
@@ -251,11 +262,16 @@ void NodeList::addSetOfNodeTypesToNodeInterestSet(const NodeSet& setOfNodeTypes)
 }
 
 void NodeList::sendDomainServerCheckIn() {
+    if (thread() != QThread::currentThread()) {
+        QMetaObject::invokeMethod(this, "sendDomainServerCheckIn", Qt::QueuedConnection);
+        return;
+    }
+
     if (_isShuttingDown) {
         qCDebug(networking) << "Refusing to send a domain-server check in while shutting down.";
         return;
     }
-    
+
     if (_publicSockAddr.isNull()) {
         // we don't know our public socket and we need to send it to the domain server
         qCDebug(networking) << "Waiting for inital public socket from STUN. Will not send domain-server check in.";

libraries/networking/src/NodeList.h

@@ -54,8 +54,8 @@ public:
     NodeType_t getOwnerType() const { return _ownerType; }
     void setOwnerType(NodeType_t ownerType) { _ownerType = ownerType; }
 
-    qint64 sendStats(const QJsonObject& statsObject, const HifiSockAddr& destination);
-    qint64 sendStatsToDomainServer(const QJsonObject& statsObject);
+    Q_INVOKABLE qint64 sendStats(QJsonObject statsObject, HifiSockAddr destination);
+    Q_INVOKABLE qint64 sendStatsToDomainServer(QJsonObject statsObject);
 
     int getNumNoReplyDomainCheckIns() const { return _numNoReplyDomainCheckIns; }
     DomainHandler& getDomainHandler() { return _domainHandler; }

libraries/networking/src/ThreadedAssignment.cpp

@@ -87,11 +87,10 @@ void ThreadedAssignment::commonInit(const QString& targetName, NodeType_t nodeTy
     connect(&nodeList->getDomainHandler(), &DomainHandler::disconnectedFromDomain, &_statsTimer, &QTimer::stop);
 }
 
-void ThreadedAssignment::addPacketStatsAndSendStatsPacket(QJsonObject &statsObject) {
+void ThreadedAssignment::addPacketStatsAndSendStatsPacket(QJsonObject statsObject) {
     auto nodeList = DependencyManager::get<NodeList>();
 
     float packetsPerSecond, bytesPerSecond;
-    // XXX can BandwidthRecorder be used for this?
     nodeList->getPacketStats(packetsPerSecond, bytesPerSecond);
     nodeList->resetPacketStats();
 

libraries/networking/src/ThreadedAssignment.h

@@ -26,7 +26,7 @@ public:
 
     void setFinished(bool isFinished);
     virtual void aboutToFinish() { };
-    void addPacketStatsAndSendStatsPacket(QJsonObject& statsObject);
+    void addPacketStatsAndSendStatsPacket(QJsonObject statsObject);
 
 public slots:
     /// threaded run of assignment

libraries/shared/src/SimpleMovingAverage.cpp

@@ -19,15 +19,29 @@ SimpleMovingAverage::SimpleMovingAverage(int numSamplesToAverage) :
     _eventDeltaAverage(0.0f),
     WEIGHTING(1.0f / numSamplesToAverage),
     ONE_MINUS_WEIGHTING(1 - WEIGHTING) {
-        
 }
 
+SimpleMovingAverage::SimpleMovingAverage(const SimpleMovingAverage& other) {
+    *this = other;
+}
+
+SimpleMovingAverage& SimpleMovingAverage::operator=(const SimpleMovingAverage& other) {
+    _numSamples = (int)other._numSamples;
+    _lastEventTimestamp = (uint64_t)other._lastEventTimestamp;
+    _average = (float)other._average;
+    _eventDeltaAverage = (float)other._eventDeltaAverage;
+    WEIGHTING = other.WEIGHTING;
+    ONE_MINUS_WEIGHTING = other.ONE_MINUS_WEIGHTING;
+    return *this;
+}
+
+
 int SimpleMovingAverage::updateAverage(float sample) {
     if (_numSamples > 0) {
         _average = (ONE_MINUS_WEIGHTING * _average) + (WEIGHTING * sample);
-        
+
         float eventDelta = (usecTimestampNow() - _lastEventTimestamp) / 1000000.0f;
-        
+
         if (_numSamples > 1) {
             _eventDeltaAverage = (ONE_MINUS_WEIGHTING * _eventDeltaAverage) +
                 (WEIGHTING * eventDelta);

libraries/shared/src/SimpleMovingAverage.h

@@ -16,27 +16,31 @@
 
 #include <mutex>
 #include <stdint.h>
+#include <atomic>
 
 class SimpleMovingAverage {
 public:
     SimpleMovingAverage(int numSamplesToAverage = 100);
-    
+    SimpleMovingAverage(const SimpleMovingAverage& other);
+    SimpleMovingAverage& operator=(const SimpleMovingAverage& other);
+
     int updateAverage(float sample);
     void reset();
-    
+
     int getSampleCount() const { return _numSamples; };
     float getAverage() const { return _average; };
     float getEventDeltaAverage() const; // returned in seconds
     float getAverageSampleValuePerSecond() const { return _average * (1.0f / getEventDeltaAverage()); }
-    
+
     uint64_t getUsecsSinceLastEvent() const;
 
+
 private:
-    int _numSamples;
-    uint64_t _lastEventTimestamp;
-    float _average;
-    float _eventDeltaAverage;
-    
+    std::atomic<int> _numSamples;
+    std::atomic<uint64_t> _lastEventTimestamp;
+    std::atomic<float> _average;
+    std::atomic<float> _eventDeltaAverage;
+
     float WEIGHTING;
     float ONE_MINUS_WEIGHTING;
 };
@@ -44,10 +48,20 @@ private:
 
 template <class T, int MAX_NUM_SAMPLES> class MovingAverage {
 public:
+    MovingAverage<T, MAX_NUM_SAMPLES>() {}
+    MovingAverage<T, MAX_NUM_SAMPLES>(const MovingAverage<T, MAX_NUM_SAMPLES>& other) {
+        *this = other;
+    }
+    MovingAverage<T, MAX_NUM_SAMPLES>& operator=(const MovingAverage<T, MAX_NUM_SAMPLES>& other) {
+        numSamples = (int)other.numSamples;
+        average = (T)other.average;
+        return *this;
+    }
+
     const float WEIGHTING = 1.0f / (float)MAX_NUM_SAMPLES;
     const float ONE_MINUS_WEIGHTING = 1.0f - WEIGHTING;
-    int numSamples{ 0 };
-    T average;
+    std::atomic<int> numSamples{ 0 };
+    std::atomic<T> average;
 
     void clear() {
         numSamples = 0;
@@ -72,7 +86,7 @@ public:
         _samples = 0;
     }
 
-    bool isAverageValid() const { 
+    bool isAverageValid() const {
         std::unique_lock<std::mutex> lock(_lock);
         return (_samples > 0);
     }
@@ -87,7 +101,7 @@ public:
         _samples++;
     }
 
-    T getAverage() const { 
+    T getAverage() const {
         std::unique_lock<std::mutex> lock(_lock);
         return _average;
     }