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.master
parent
af35391114
commit
bf8aea5059
|
|
@ -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({});
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in New Issue