diff options
| -rw-r--r-- | indra/llmessage/llcoproceduremanager.cpp | 31 |
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({}); } |