DRTVWR-476: Use LLThreadSafeQueue, not boost::fibers::buffered_channel.

We've observed buffered_channel::try_push() hanging, which seems very odd. Try
our own LLThreadSafeQueue instead.

1 files changed, 11 insertions, 20 deletions

diff --git a/indra/llmessage/llcoproceduremanager.cpp b/indra/llmessage/llcoproceduremanager.cpp
index 1c925b7eea..712cab5b19 100644
--- a/indra/llmessage/llcoproceduremanager.cpp
+++ b/indra/llmessage/llcoproceduremanager.cpp
@@ -33,7 +33,7 @@
 
 #include <chrono>
 
-#include <boost/fiber/buffered_channel.hpp>
+#include "llthreadsafequeue.h"
 
 #include "llexception.h"
 #include "stringize.h"
@@ -105,10 +105,7 @@ private:
         CoProcedure_t mProc;
     };
 
-    // we use a buffered_channel here rather than unbuffered_channel since we want to be able to 
-    // push values without blocking,even if there's currently no one calling a pop operation (due to
-    // fiber running right now)
-    typedef boost::fibers::buffered_channel<QueuedCoproc::ptr_t>  CoprocQueue_t;
+    typedef LLThreadSafeQueue<QueuedCoproc::ptr_t>  CoprocQueue_t;
     // Use shared_ptr to control the lifespan of our CoprocQueue_t instance
     // because the consuming coroutine might outlive this LLCoprocedurePool
     // instance.
@@ -289,7 +286,7 @@ LLCoprocedurePool::LLCoprocedurePool(const std::string &poolName, size_t size):
                                       << LL_ENDL;
                 // This should ensure that all waiting coprocedures in this
                 // pool will wake up and terminate.
-                pendingCoprocs->close();
+                pendingCoprocs->pushFront({});
             }
             return false;
         });
@@ -319,14 +316,13 @@ LLUUID LLCoprocedurePool::enqueueCoprocedure(const std::string &name, LLCoproced
     LLUUID id(LLUUID::generateNewID());
 
     LL_INFOS("CoProcMgr") << "Coprocedure(" << name << ") enqueuing with id=" << id.asString() << " in pool \"" << mPoolName << "\" at " << mPending << LL_ENDL;
-    auto pushed = mPendingCoprocs->try_push(boost::make_shared<QueuedCoproc>(name, id, proc));
-    // We don't really have a lot of good options if try_push() failed,
+    auto pushed = mPendingCoprocs->tryPushFront(boost::make_shared<QueuedCoproc>(name, id, proc));
+    // We don't really have a lot of good options if tryPushFront() failed,
     // perhaps because the consuming coroutine is gummed up or something. This
     // method is probably called from code called by mainloop. If we toss an
     // llcoro::suspend() call here, we'll circle back for another mainloop
     // iteration, possibly resulting in being re-entered here. Let's avoid that.
-    LL_ERRS_IF(pushed != boost::fibers::channel_op_status::success, "CoProcMgr")
-        << "Enqueue failed because queue is " << int(pushed) << LL_ENDL;
+    LL_ERRS_IF(! pushed, "CoProcMgr") << "Enqueue failed" << LL_ENDL;
     ++mPending;
 
     return id;
@@ -338,23 +334,18 @@ void LLCoprocedurePool::coprocedureInvokerCoro(
     LLCoreHttpUtil::HttpCoroutineAdapter::ptr_t httpAdapter)
 {
     QueuedCoproc::ptr_t coproc;
-    boost::fibers::channel_op_status status;
     for (;;)
     {
         {
-            LLCoros::TempStatus st("waiting for work for 10s");
-            status = pendingCoprocs->pop_wait_for(coproc, std::chrono::seconds(10));
+            LLCoros::TempStatus st("waiting for work");
+            coproc = pendingCoprocs->popBack();
         }
-        if (status == boost::fibers::channel_op_status::closed)
+        if (! coproc)
         {
+            // close() pushes an empty pointer to signal done
             break;
         }
 
-        if(status == boost::fibers::channel_op_status::timeout)
-        {
-            LL_INFOS_ONCE() << "pool '" << mPoolName << "' stalled." << LL_ENDL;
-            continue;
-        }
         // we actually popped an item
         --mPending;
 
@@ -384,5 +375,5 @@ void LLCoprocedurePool::coprocedureInvokerCoro(
 
 void LLCoprocedurePool::close()
 {
-    mPendingCoprocs->close();
+    mPendingCoprocs->pushFront({});
 }