Merge branch 'development' into linux_ar

Code/Framework/AzCore/AzCore/Asset/AssetCommon.h

@@ -556,16 +556,24 @@ namespace AZ
                     Asset<AssetData> assetData(AssetInternal::GetAssetData(actualId, AZ::Data::AssetLoadBehavior::Default));
                     if (assetData)
                     {
-                        auto curStatus = assetData->GetStatus();
+                        auto isReady = assetData->GetStatus() == AssetData::AssetStatus::Ready;
                         bool isError = assetData->IsError();
-                        connectLock.unlock();
-                        if (curStatus == AssetData::AssetStatus::Ready)
+
+                        if (isReady || isError)
                         {
-                            handler->OnAssetReady(assetData);
-                        }
-                        else if (isError)
-                        {
-                            handler->OnAssetError(assetData);
+                            connectLock.unlock();
+
+                            if (isReady)
+                            {
+                                handler->OnAssetReady(assetData);
+                            }
+                            else if (isError)
+                            {
+                                handler->OnAssetError(assetData);
+                            }
+
+                            // Lock the mutex again since some destructors will be modifying the context afterwards
+                            connectLock.lock();
                         }
                     }
                 }

Code/Framework/AzCore/AzCore/Asset/AssetManager.cpp

@@ -1677,9 +1677,13 @@ namespace AZ
                 // they will trigger a ReleaseAsset call sometime after the AssetManager has begun to shut down, which can lead to
                 // race conditions.
 
+                // Make sure the streamer request is removed first before the asset is released
+                // If the asset is released first it could lead to a race condition where another thread starts loading the asset
+                // again and attempts to add a new streamer request with the same ID before the old one has been removed, causing
+                // that load request to fail
+                RemoveActiveStreamerRequest(assetId);
                 weakAsset = {};
                 loadingAsset.Reset();
-                RemoveActiveStreamerRequest(assetId);
             };
 
             auto&& [deadline, priority] = GetEffectiveDeadlineAndPriority(*handler, asset.GetType(), loadParams);

Code/Framework/AzCore/Tests/Asset/AssetManagerLoadingTests.cpp

@@ -652,7 +652,7 @@ namespace UnitTest
             threads.emplace_back([this, &threadCount, &cv, assetUuid]() {
                 bool checkLoaded = true;
 
-                for (int i = 0; i < 5000; i++)
+                for (int i = 0; i < 1000; i++)
                 {
                     Asset<AssetWithAssetReference> asset1 =
                         m_testAssetManager->GetAsset(assetUuid, azrtti_typeid<AssetWithAssetReference>(), AZ::Data::AssetLoadBehavior::PreLoad);
@@ -678,7 +678,7 @@ namespace UnitTest
         while (threadCount > 0 && !timedOut)
         {
             AZStd::unique_lock<AZStd::mutex> lock(mutex);
-            timedOut = (AZStd::cv_status::timeout == cv.wait_until(lock, AZStd::chrono::system_clock::now() + DefaultTimeoutSeconds * 20000));
+            timedOut = (AZStd::cv_status::timeout == cv.wait_until(lock, AZStd::chrono::system_clock::now() + DefaultTimeoutSeconds));
         }
 
         ASSERT_EQ(threadCount, 0) << "Thread count is non-zero, a thread has likely deadlocked.  Test will not shut down cleanly.";
@@ -1190,7 +1190,7 @@ namespace UnitTest
 #if AZ_TRAIT_DISABLE_FAILED_ASSET_MANAGER_TESTS
     TEST_F(AssetJobsFloodTest, DISABLED_ContainerFilterTest_ContainersWithAndWithoutFiltering_Success)
 #else
-    TEST_F(AssetJobsFloodTest, DISABLED_ContainerFilterTest_ContainersWithAndWithoutFiltering_Success)
+    TEST_F(AssetJobsFloodTest, ContainerFilterTest_ContainersWithAndWithoutFiltering_Success)
 #endif // !AZ_TRAIT_DISABLE_FAILED_ASSET_MANAGER_TESTS
     {
         m_assetHandlerAndCatalog->AssetCatalogRequestBus::Handler::BusConnect();

Gems/Atom/RHI/Code/Include/Atom/RHI/CommandQueue.h

@@ -72,6 +72,8 @@ namespace AZ
             AZStd::mutex m_workQueueMutex;
             AZStd::queue<Command> m_workQueue;
             AZStd::condition_variable m_workQueueCondition;
+            AZStd::mutex m_flushCommandsMutex;
+            AZStd::condition_variable m_flushCommandsCondition;
             AZStd::atomic_bool m_isWorkQueueEmpty;
             AZStd::atomic_bool m_isQuitting;
         };

Gems/Atom/RHI/Code/Source/RHI/CommandQueue.cpp

@@ -71,6 +71,7 @@ namespace AZ
             {
                 m_isQuitting = true;
                 m_workQueueCondition.notify_all();
+                m_flushCommandsCondition.notify_all();
                 if (m_thread.joinable())
                 {
                    m_thread.join();
@@ -102,9 +103,10 @@ namespace AZ
         void CommandQueue::FlushCommands()
         {
             AZ_PROFILE_SCOPE(RHI, "CommandQueue: FlushCommands");
-            while (!m_isWorkQueueEmpty && !m_isQuitting)
+            AZStd::unique_lock<AZStd::mutex> lock(m_flushCommandsMutex);
+            if (!m_isWorkQueueEmpty && !m_isQuitting)
             {
-                AZStd::this_thread::yield();
+                m_flushCommandsCondition.wait(lock, [this]() { return m_isWorkQueueEmpty.load() || m_isQuitting.load(); });
             }
         }
         
@@ -119,7 +121,11 @@ namespace AZ
                     
                     if (m_workQueue.empty())
                     {
-                        m_isWorkQueueEmpty = true;
+                        {
+                            AZStd::unique_lock<AZStd::mutex> flushCommandsLock(m_flushCommandsMutex);
+                            m_isWorkQueueEmpty = true;
+                            m_flushCommandsCondition.notify_all();
+                        }
                         m_workQueueCondition.wait(lock, [this]() { return !m_workQueue.empty() || m_isQuitting; });
                     }