Use SPSC queue in image pool workers for deep request pipelining

indra/deps/CMakeLists.txt

@@ -54,6 +54,11 @@ FetchContent_Declare(
   GIT_REPOSITORY https://git.alchemyviewer.org/alchemy/mirrors/abseil-cpp.git
   GIT_TAG        9a7e447c511dae7276ab65fde4d04f6ed52b39c9
   )
+FetchContent_Declare(
+  readerwriterqueue
+  GIT_REPOSITORY    https://github.com/cameron314/readerwriterqueue
+  GIT_TAG           v1.0.5
+)
 
 # This is a hack because absl has dumb cmake
 set(OLD_BUILD_TEST ${BUILD_TESTING})
@@ -94,5 +99,7 @@ set(JSON_Install OFF CACHE INTERNAL "")
 set(JSON_BuildTests OFF CACHE INTERNAL "")
 FetchContent_MakeAvailable(nlohmann_json)
 
+FetchContent_MakeAvailable(readerwriterqueue)
+
 unset(CMAKE_FOLDER)
 unset(CMAKE_POSITION_INDEPENDENT_CODE)

indra/llimage/CMakeLists.txt

@@ -79,6 +79,7 @@ target_link_libraries(llimage
     ${JPEG_LIBRARIES}
     ${PNG_LIBRARIES}
     ${ZLIB_LIBRARIES}
+    readerwriterqueue
     )
 
 if(${CMAKE_VERSION} VERSION_GREATER "3.15.0") 

indra/llimage/llimageworker.cpp

@@ -28,54 +28,49 @@
 
 #include "llimageworker.h"
 #include "llimagedxt.h"
+#include <readerwriterqueue.h>
 
 std::atomic< U32 > sImageThreads;
 
 class PoolWorkerThread : public LLThread
 {
 public:
-	PoolWorkerThread(std::string name) : LLThread(name),
-		mCurrentRequest(NULL)
+	PoolWorkerThread(std::string name) : LLThread(name), mRequestQueue(30)
 	{
 	}
+
 	virtual void run()
 	{
 		while (!isQuitting())
 		{
-			auto *pReq = mCurrentRequest.exchange(nullptr);
-
-			if (pReq)
-				pReq->processRequestIntern();
-			checkPause();
+            checkPause();
+
+			LLImageDecodeThread::ImageRequest* req  = nullptr;
+            while (!isQuitting() && mRequestQueue.try_dequeue(req))
+            {
+                if (req)
+                {
+                    req->processRequestIntern();
+                }
+            }
 		}
 	}
-	bool isBusy()
-	{
-		auto *pReq = mCurrentRequest.load();
-		if (!pReq)
-			return false;
-
-		auto status = pReq->getStatus();
-
-		return status  == LLQueuedThread::STATUS_QUEUED || status == LLQueuedThread::STATUS_INPROGRESS;
-	}
 
 	bool runCondition()
-	{
-		return mCurrentRequest != NULL;
+    {
+        return mRequestQueue.size_approx() > 0;
 	}
 
 	bool setRequest(LLImageDecodeThread::ImageRequest* req)
 	{
-		LLImageDecodeThread::ImageRequest* pOld{ nullptr };
-		bool bSuccess = mCurrentRequest.compare_exchange_strong(pOld, req);
+        bool bSuccess = mRequestQueue.try_enqueue(req);
 		wake();
 
 		return bSuccess;
 	}
 
 private:
-	std::atomic< LLImageDecodeThread::ImageRequest * > mCurrentRequest;
+    moodycamel::ReaderWriterQueue<LLImageDecodeThread::ImageRequest*> mRequestQueue;
 };
 
 //----------------------------------------------------------------------------
@@ -344,7 +339,7 @@ bool LLImageDecodeThread::enqueRequest(ImageRequest * req)
             mLastPoolAllocation = 0;
         }
         auto& thread = mThreadPool[mLastPoolAllocation++];
-        if (!thread->isBusy() && thread->setRequest(req))
+        if (thread->setRequest(req))
         {
             return true;
         }