DRTVWR-476: Use LLThreadSafeQueue::close() to shut down coprocs.

The tactic of pushing an empty QueuedCoproc::ptr_t to signal coprocedure close
only works for LLCoprocedurePools with a single coprocedure (e.g. "Upload" and
"AIS"). Only one coprocedureInvokerCoro() coroutine will pop that empty
pointer and shut down properly -- the rest will continue waiting indefinitely.

Rather than pushing some number of empty pointers, hopefully enough to notify
all consumer coroutines, close() the queue. That will notify as many consumers
as there may be.

That means catching LLThreadSafeQueueInterrupt from popBack(), instead of
detecting empty pointer.

Also, if a queued coprocedure throws an exception, coprocedureInvokerCoro()
logs it as before -- but instead of rethrowing it, the coroutine now loops
back to wait for more work. Otherwise, the number of coroutines servicing the
queue dwindles.

1 files changed, 7 insertions, 6 deletions

diff --git a/indra/llmessage/llcoproceduremanager.cpp b/indra/llmessage/llcoproceduremanager.cpp
index c1e53ea278..d252c0e4b0 100644
--- a/indra/llmessage/llcoproceduremanager.cpp
+++ b/indra/llmessage/llcoproceduremanager.cpp
@@ -291,7 +291,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->pushFront({});
+                pendingCoprocs->close();
             }
             return false;
         });
@@ -323,7 +323,7 @@ LLUUID LLCoprocedurePool::enqueueCoprocedure(const std::string &name, LLCoproced
     LL_INFOS("CoProcMgr") << "Coprocedure(" << name << ") enqueuing with id=" << id.asString() << " in pool \"" << mPoolName << "\" at " << mPending << LL_ENDL;
     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
+    // perhaps because the consuming coroutines are 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.
@@ -341,13 +341,14 @@ void LLCoprocedurePool::coprocedureInvokerCoro(
     QueuedCoproc::ptr_t coproc;
     for (;;)
     {
+        try
         {
             LLCoros::TempStatus st("waiting for work");
             coproc = pendingCoprocs->popBack();
         }
-        if (! coproc)
+        catch (const LLThreadSafeQueueError&)
         {
-            // close() pushes an empty pointer to signal done
+            // queue is closed
             break;
         }
 
@@ -369,7 +370,7 @@ void LLCoprocedurePool::coprocedureInvokerCoro(
                                               << ") in pool '" << mPoolName << "'"));
             // must NOT omit this or we deplete the pool
             mActiveCoprocs.erase(itActive);
-            throw;
+            continue;
         }
 
         LL_DEBUGS("CoProcMgr") << "Finished coprocedure(" << coproc->mName << ")" << " in pool \"" << mPoolName << "\"" << LL_ENDL;
@@ -380,5 +381,5 @@ void LLCoprocedurePool::coprocedureInvokerCoro(
 
 void LLCoprocedurePool::close()
 {
-    mPendingCoprocs->pushFront({});
+    mPendingCoprocs->close();
 }