From b6841d75c2f259c84d5ab6b012bd2ae37d985451 Mon Sep 17 00:00:00 2001 From: Dave Parks Date: Fri, 15 Apr 2022 19:02:07 -0500 Subject: SL-17219 WIP - Texture pipeline overhaul --- indra/llcommon/workqueue.h | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) (limited to 'indra/llcommon/workqueue.h') diff --git a/indra/llcommon/workqueue.h b/indra/llcommon/workqueue.h index 96574a18b9..46f7363830 100644 --- a/indra/llcommon/workqueue.h +++ b/indra/llcommon/workqueue.h @@ -161,10 +161,16 @@ namespace LL void postEvery(const std::chrono::duration& interval, CALLABLE&& callable); + template + bool tryPost(const TimePoint& time, CALLABLE&& callable) + { + return mQueue.tryPush(TimedWork(time, std::move(callable))); + } + template bool tryPost(CALLABLE&& callable) { - return mQueue.tryPush(TimedWork(TimePoint::clock::now(), std::move(callable))); + return mQueue.tryPost(TimePoint::clock::now(), std::move(callable)); } /*------------------------- handshake API --------------------------*/ -- cgit v1.2.3 From 72ddfbd76ef3152c86e9b0b4331919d15d6a3d2a Mon Sep 17 00:00:00 2001 From: Nat Goodspeed Date: Tue, 10 May 2022 12:18:27 -0400 Subject: SL-17219: WorkQueue::tryPost() must call ThreadSafeSchedule::tryPush(). We inadvertently changed tryPost() to call ThreadSafeSchedule::tryPost(), which doesn't exist. --- indra/llcommon/workqueue.h | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) (limited to 'indra/llcommon/workqueue.h') diff --git a/indra/llcommon/workqueue.h b/indra/llcommon/workqueue.h index 46f7363830..784327f929 100644 --- a/indra/llcommon/workqueue.h +++ b/indra/llcommon/workqueue.h @@ -170,7 +170,7 @@ namespace LL template bool tryPost(CALLABLE&& callable) { - return mQueue.tryPost(TimePoint::clock::now(), std::move(callable)); + return mQueue.tryPush(TimePoint::clock::now(), std::move(callable)); } /*------------------------- handshake API --------------------------*/ -- cgit v1.2.3 From fc424a0db90fd2d2e44e85a19750ad6eaa57b28a Mon Sep 17 00:00:00 2001 From: Nat Goodspeed Date: Fri, 9 Dec 2022 13:21:45 -0500 Subject: SL-18809: Add WorkSchedule; remove timestamps from WorkQueue. For work queues that don't need timestamped tasks, eliminate the overhead of a priority queue ordered by timestamp. Timestamped task support moves to WorkSchedule. WorkQueue is a simpler queue that just waits for work. Both WorkQueue and WorkSchedule can be accessed via new WorkQueueBase API. Of course the WorkQueueBase API doesn't deal with timestamps, but a WorkSchedule can be accessed directly to post timestamped tasks and then handled normally (e.g. by ThreadPool) to run them. Most ThreadPool functionality migrates to new ThreadPoolBase class, with template subclass ThreadPoolUsing or ThreadPoolUsing depending on need. ThreadPool is now an alias for ThreadPoolUsing. Importantly, ThreadPoolUsing::getQueue() delivers a reference to the specific queue subclass type, so you can post timestamped tasks on a queue retrieved from ThreadPoolUsing::getQueue(). Since ThreadPool is no longer a simple class but an alias for a particular template specialization, introduce threadpool_fwd.h to forward-declare it. Recast workqueue_test.cpp to exercise WorkSchedule, since some of the tests are time-based. A future todo would be to exercise each applicable test with both WorkQueue and WorkSchedule. --- indra/llcommon/workqueue.h | 416 ++++++++++++++++++++++++++------------------- 1 file changed, 244 insertions(+), 172 deletions(-) (limited to 'indra/llcommon/workqueue.h') diff --git a/indra/llcommon/workqueue.h b/indra/llcommon/workqueue.h index 28a0b5e040..eea8886a7a 100644 --- a/indra/llcommon/workqueue.h +++ b/indra/llcommon/workqueue.h @@ -15,6 +15,7 @@ #include "llcoros.h" #include "llexception.h" #include "llinstancetracker.h" +#include "llinstancetrackersubclass.h" #include "threadsafeschedule.h" #include #include // std::current_exception @@ -23,27 +24,23 @@ namespace LL { + +/***************************************************************************** +* WorkQueueBase: API for WorkQueue and WorkSchedule +*****************************************************************************/ /** * A typical WorkQueue has a string name that can be used to find it. */ - class WorkQueue: public LLInstanceTracker + class WorkQueueBase: public LLInstanceTracker { private: - using super = LLInstanceTracker; + using super = LLInstanceTracker; public: using Work = std::function; - - private: - using Queue = ThreadSafeSchedule; - // helper for postEvery() - template - class BackJack; - - public: - using TimePoint = Queue::TimePoint; - using TimedWork = Queue::TimeTuple; - using Closed = Queue::Closed; + using Closed = LLThreadSafeQueueInterrupt; + // for runFor() + using TimePoint = std::chrono::steady_clock::time_point; struct Error: public LLException { @@ -51,18 +48,18 @@ namespace LL }; /** - * You may omit the WorkQueue name, in which case a unique name is + * You may omit the WorkQueueBase name, in which case a unique name is * synthesized; for practical purposes that makes it anonymous. */ - WorkQueue(const std::string& name = std::string(), size_t capacity=1024); + WorkQueueBase(const std::string& name); /** * Since the point of WorkQueue is to pass work to some other worker - * thread(s) asynchronously, it's important that the WorkQueue continue - * to exist until the worker thread(s) have drained it. To communicate - * that it's time for them to quit, close() the queue. + * thread(s) asynchronously, it's important that it continue to exist + * until the worker thread(s) have drained it. To communicate that + * it's time for them to quit, close() the queue. */ - void close(); + virtual void close() = 0; /** * WorkQueue supports multiple producers and multiple consumers. In @@ -78,158 +75,60 @@ namespace LL * * If you're the only consumer, noticing that size() > 0 is * meaningful. */ - size_t size(); + virtual size_t size() = 0; /// producer end: are we prevented from pushing any additional items? - bool isClosed(); + virtual bool isClosed() = 0; /// consumer end: are we done, is the queue entirely drained? - bool done(); + virtual bool done() = 0; /*---------------------- fire and forget API -----------------------*/ - /// fire-and-forget, but at a particular (future?) time - template - void post(const TimePoint& time, CALLABLE&& callable) - { - // Defer reifying an arbitrary CALLABLE until we hit this or - // postIfOpen(). All other methods should accept CALLABLEs of - // arbitrary type to avoid multiple levels of std::function - // indirection. - mQueue.push(TimedWork(time, std::move(callable))); - } - /// fire-and-forget - template - void post(CALLABLE&& callable) - { - // We use TimePoint::clock::now() instead of TimePoint's - // representation of the epoch because this WorkQueue may contain - // a mix of past-due TimedWork items and TimedWork items scheduled - // for the future. Sift this new item into the correct place. - post(TimePoint::clock::now(), std::move(callable)); - } - - /** - * post work for a particular time, unless the queue is closed before - * we can post - */ - template - bool postIfOpen(const TimePoint& time, CALLABLE&& callable) - { - // Defer reifying an arbitrary CALLABLE until we hit this or - // post(). All other methods should accept CALLABLEs of arbitrary - // type to avoid multiple levels of std::function indirection. - return mQueue.pushIfOpen(TimedWork(time, std::move(callable))); - } + virtual void post(const Work&) = 0; /** * post work, unless the queue is closed before we can post */ - template - bool postIfOpen(CALLABLE&& callable) - { - return postIfOpen(TimePoint::clock::now(), std::move(callable)); - } + virtual bool postIfOpen(const Work&) = 0; /** - * Post work to be run at a specified time to another WorkQueue, which - * may or may not still exist and be open. Return true if we were able - * to post. + * post work, unless the queue is full */ - template - static bool postMaybe(weak_t target, const TimePoint& time, CALLABLE&& callable); + virtual bool tryPost(const Work&) = 0; /** * Post work to another WorkQueue, which may or may not still exist - * and be open. Return true if we were able to post. - */ - template - static bool postMaybe(weak_t target, CALLABLE&& callable) - { - return postMaybe(target, TimePoint::clock::now(), - std::forward(callable)); - } - - /** - * Launch a callable returning bool that will trigger repeatedly at - * specified interval, until the callable returns false. - * - * If you need to signal that callable from outside, DO NOT bind a - * reference to a simple bool! That's not thread-safe. Instead, bind - * an LLCond variant, e.g. LLOneShotCond or LLBoolCond. + * and be open. Support any post() overload. Return true if we were + * able to post. */ - template - void postEvery(const std::chrono::duration& interval, - CALLABLE&& callable); - - template - bool tryPost(const TimePoint& time, CALLABLE&& callable) - { - return mQueue.tryPush(TimedWork(time, std::move(callable))); - } - - template - bool tryPost(CALLABLE&& callable) - { - return mQueue.tryPush(TimePoint::clock::now(), std::move(callable)); - } + template + static bool postMaybe(weak_t target, ARGS&&... args); /*------------------------- handshake API --------------------------*/ - /** - * Post work to another WorkQueue to be run at a specified time, - * requesting a specific callback to be run on this WorkQueue on - * completion. - * - * Returns true if able to post, false if the other WorkQueue is - * inaccessible. - */ - // Apparently some Microsoft header file defines a macro CALLBACK? The - // natural template argument name CALLBACK produces very weird Visual - // Studio compile errors that seem utterly unrelated to this source - // code. - template - bool postTo(weak_t target, - const TimePoint& time, CALLABLE&& callable, FOLLOWUP&& callback); - /** * Post work to another WorkQueue, requesting a specific callback to - * be run on this WorkQueue on completion. + * be run on this WorkQueue on completion. Optional final argument is + * TimePoint for WorkSchedule. * * Returns true if able to post, false if the other WorkQueue is * inaccessible. */ - template - bool postTo(weak_t target, CALLABLE&& callable, FOLLOWUP&& callback) - { - return postTo(target, TimePoint::clock::now(), - std::move(callable), std::move(callback)); - } - - /** - * Post work to another WorkQueue to be run at a specified time, - * blocking the calling coroutine until then, returning the result to - * caller on completion. - * - * In general, we assume that each thread's default coroutine is busy - * servicing its WorkQueue or whatever. To try to prevent mistakes, we - * forbid calling waitForResult() from a thread's default coroutine. - */ - template - auto waitForResult(const TimePoint& time, CALLABLE&& callable); + template + bool postTo(weak_t target, CALLABLE&& callable, FOLLOWUP&& callback, + ARGS&&... args); /** * Post work to another WorkQueue, blocking the calling coroutine - * until then, returning the result to caller on completion. + * until then, returning the result to caller on completion. Optional + * final argument is TimePoint for WorkSchedule. * * In general, we assume that each thread's default coroutine is busy * servicing its WorkQueue or whatever. To try to prevent mistakes, we * forbid calling waitForResult() from a thread's default coroutine. */ - template - auto waitForResult(CALLABLE&& callable) - { - return waitForResult(TimePoint::clock::now(), std::move(callable)); - } + template + auto waitForResult(CALLABLE&& callable, ARGS&&... args); /*--------------------------- worker API ---------------------------*/ @@ -276,7 +175,7 @@ namespace LL */ bool runUntil(const TimePoint& until); - private: + protected: template static auto makeReplyLambda(CALLABLE&& callable, FOLLOWUP&& callback); /// general case: arbitrary C++ return type @@ -296,13 +195,179 @@ namespace LL static void checkCoroutine(const std::string& method); static void error(const std::string& msg); static std::string makeName(const std::string& name); - void callWork(const Queue::DataTuple& work); void callWork(const Work& work); + + private: + virtual Work pop_() = 0; + virtual bool tryPop_(Work&) = 0; + }; + +/***************************************************************************** +* WorkQueue: no timestamped task support +*****************************************************************************/ + class WorkQueue: public LLInstanceTrackerSubclass + { + private: + using super = LLInstanceTrackerSubclass; + + public: + /** + * You may omit the WorkQueue name, in which case a unique name is + * synthesized; for practical purposes that makes it anonymous. + */ + WorkQueue(const std::string& name = std::string(), size_t capacity=1024); + + /** + * Since the point of WorkQueue is to pass work to some other worker + * thread(s) asynchronously, it's important that it continue to exist + * until the worker thread(s) have drained it. To communicate that + * it's time for them to quit, close() the queue. + */ + void close() override; + + /** + * WorkQueue supports multiple producers and multiple consumers. In + * the general case it's misleading to test size(), since any other + * thread might change it the nanosecond the lock is released. On that + * basis, some might argue against publishing a size() method at all. + * + * But there are two specific cases in which a test based on size() + * might be reasonable: + * + * * If you're the only producer, noticing that size() == 0 is + * meaningful. + * * If you're the only consumer, noticing that size() > 0 is + * meaningful. + */ + size_t size() override; + /// producer end: are we prevented from pushing any additional items? + bool isClosed() override; + /// consumer end: are we done, is the queue entirely drained? + bool done() override; + + /*---------------------- fire and forget API -----------------------*/ + + /// fire-and-forget + void post(const Work&) override; + + /** + * post work, unless the queue is closed before we can post + */ + bool postIfOpen(const Work&) override; + + /** + * post work, unless the queue is full + */ + bool tryPost(const Work&) override; + + private: + using Queue = LLThreadSafeQueue; Queue mQueue; + + Work pop_() override; + bool tryPop_(Work&) override; + }; + +/***************************************************************************** +* WorkSchedule: add support for timestamped tasks +*****************************************************************************/ + class WorkSchedule: public LLInstanceTrackerSubclass + { + private: + using super = LLInstanceTrackerSubclass; + using Queue = ThreadSafeSchedule; + // helper for postEvery() + template + class BackJack; + + public: + using TimePoint = Queue::TimePoint; + using TimedWork = Queue::TimeTuple; + + /** + * You may omit the WorkSchedule name, in which case a unique name is + * synthesized; for practical purposes that makes it anonymous. + */ + WorkSchedule(const std::string& name = std::string(), size_t capacity=1024); + + /** + * Since the point of WorkSchedule is to pass work to some other worker + * thread(s) asynchronously, it's important that the WorkSchedule continue + * to exist until the worker thread(s) have drained it. To communicate + * that it's time for them to quit, close() the queue. + */ + void close() override; + + /** + * WorkSchedule supports multiple producers and multiple consumers. In + * the general case it's misleading to test size(), since any other + * thread might change it the nanosecond the lock is released. On that + * basis, some might argue against publishing a size() method at all. + * + * But there are two specific cases in which a test based on size() + * might be reasonable: + * + * * If you're the only producer, noticing that size() == 0 is + * meaningful. + * * If you're the only consumer, noticing that size() > 0 is + * meaningful. + */ + size_t size() override; + /// producer end: are we prevented from pushing any additional items? + bool isClosed() override; + /// consumer end: are we done, is the queue entirely drained? + bool done() override; + + /*---------------------- fire and forget API -----------------------*/ + + /// fire-and-forget + void post(const Work& callable) override; + + /// fire-and-forget, but at a particular (future?) time + void post(const Work& callable, const TimePoint& time); + + /** + * post work, unless the queue is closed before we can post + */ + bool postIfOpen(const Work& callable) override; + + /** + * post work for a particular time, unless the queue is closed before + * we can post + */ + bool postIfOpen(const Work& callable, const TimePoint& time); + + /** + * post work, unless the queue is full + */ + bool tryPost(const Work& callable) override; + + /** + * post work for a particular time, unless the queue is full + */ + bool tryPost(const Work& callable, const TimePoint& time); + + /** + * Launch a callable returning bool that will trigger repeatedly at + * specified interval, until the callable returns false. + * + * If you need to signal that callable from outside, DO NOT bind a + * reference to a simple bool! That's not thread-safe. Instead, bind + * an LLCond variant, e.g. LLOneShotCond or LLBoolCond. + */ + template + void postEvery(const std::chrono::duration& interval, + CALLABLE&& callable); + + private: + Queue mQueue; + + Work pop_() override; + bool tryPop_(Work&) override; }; /** - * BackJack is, in effect, a hand-rolled lambda, binding a WorkQueue, a + * BackJack is, in effect, a hand-rolled lambda, binding a WorkSchedule, a * CALLABLE that returns bool, a TimePoint and an interval at which to * relaunch it. As long as the callable continues returning true, BackJack * keeps resubmitting it to the target WorkQueue. @@ -311,7 +376,7 @@ namespace LL // class method gets its own 'this' pointer -- which we need to resubmit // the whole BackJack callable. template - class WorkQueue::BackJack + class WorkSchedule::BackJack { public: // bind the desired data @@ -325,9 +390,10 @@ namespace LL mCallable(std::move(callable)) {} - // Call by target WorkQueue -- note that although WE require a - // callable returning bool, WorkQueue wants a void callable. We - // consume the bool. + // This operator() method, called by target WorkSchedule, is what + // makes this object a Work item. Although WE require a callable + // returning bool, WorkSchedule wants a void callable. We consume the + // bool. void operator()() { // If mCallable() throws an exception, don't catch it here: if it @@ -343,7 +409,7 @@ namespace LL // register our intent to fire at exact mIntervals. mStart += mInterval; - // We're being called at this moment by the target WorkQueue. + // We're being called at this moment by the target WorkSchedule. // Assume it still exists, rather than checking the result of // lock(). // Resubmit the whole *this callable: that's why we're a class @@ -353,7 +419,8 @@ namespace LL // moved-from. try { - mTarget.lock()->post(mStart, std::move(*this)); + std::dynamic_pointer_cast(mTarget.lock())-> + post(std::move(*this), mStart); } catch (const Closed&) { @@ -370,8 +437,8 @@ namespace LL }; template - void WorkQueue::postEvery(const std::chrono::duration& interval, - CALLABLE&& callable) + void WorkSchedule::postEvery(const std::chrono::duration& interval, + CALLABLE&& callable) { if (interval.count() <= 0) { @@ -394,7 +461,7 @@ namespace LL /// general case: arbitrary C++ return type template - struct WorkQueue::MakeReplyLambda + struct WorkQueueBase::MakeReplyLambda { auto operator()(CALLABLE&& callable, FOLLOWUP&& callback) { @@ -415,7 +482,7 @@ namespace LL /// specialize for CALLABLE returning void template - struct WorkQueue::MakeReplyLambda + struct WorkQueueBase::MakeReplyLambda { auto operator()(CALLABLE&& callable, FOLLOWUP&& callback) { @@ -427,16 +494,16 @@ namespace LL }; template - auto WorkQueue::makeReplyLambda(CALLABLE&& callable, FOLLOWUP&& callback) + auto WorkQueueBase::makeReplyLambda(CALLABLE&& callable, FOLLOWUP&& callback) { return MakeReplyLambda(callable)())>() (std::move(callable), std::move(callback)); } - template - bool WorkQueue::postTo(weak_t target, - const TimePoint& time, CALLABLE&& callable, FOLLOWUP&& callback) + template + bool WorkQueueBase::postTo(weak_t target, CALLABLE&& callable, FOLLOWUP&& callback, + ARGS&&... args) { LL_PROFILE_ZONE_SCOPED; // We're being asked to post to the WorkQueue at target. @@ -450,12 +517,11 @@ namespace LL // lambda that packages our callable, our callback and a weak_ptr // to this originating WorkQueue. tptr->post( - time, [reply = super::getWeak(), callable = std::move(callable), callback = std::move(callback)] - () - mutable { + () mutable + { // Use postMaybe() below in case this originating WorkQueue // has been closed or destroyed. Remember, the outer lambda is // now running on a thread servicing the target WorkQueue, and @@ -478,14 +544,16 @@ namespace LL // originating WorkQueue. Once there, rethrow it. [exc = std::current_exception()](){ std::rethrow_exception(exc); }); } - }); + }, + // if caller passed a TimePoint, pass it along to post() + std::forward(args)...); // looks like we were able to post() return true; } - template - bool WorkQueue::postMaybe(weak_t target, const TimePoint& time, CALLABLE&& callable) + template + bool WorkQueueBase::postMaybe(weak_t target, ARGS&&... args) { LL_PROFILE_ZONE_SCOPED; // target is a weak_ptr: have to lock it to check it @@ -494,7 +562,7 @@ namespace LL { try { - tptr->post(time, std::forward(callable)); + tptr->post(std::forward(args)...); // we were able to post() return true; } @@ -509,13 +577,13 @@ namespace LL /// general case: arbitrary C++ return type template - struct WorkQueue::WaitForResult + struct WorkQueueBase::WaitForResult { - auto operator()(WorkQueue* self, const TimePoint& time, CALLABLE&& callable) + template + auto operator()(WorkQueueBase* self, CALLABLE&& callable, ARGS&&... args) { LLCoros::Promise promise; self->post( - time, // We dare to bind a reference to Promise because it's // specifically designed for cross-thread communication. [&promise, callable = std::move(callable)]() @@ -529,7 +597,9 @@ namespace LL { promise.set_exception(std::current_exception()); } - }); + }, + // if caller passed a TimePoint, pass it to post() + std::forward(args)...); auto future{ LLCoros::getFuture(promise) }; // now, on the calling thread, wait for that result LLCoros::TempStatus st("waiting for WorkQueue::waitForResult()"); @@ -539,13 +609,13 @@ namespace LL /// specialize for CALLABLE returning void template - struct WorkQueue::WaitForResult + struct WorkQueueBase::WaitForResult { - void operator()(WorkQueue* self, const TimePoint& time, CALLABLE&& callable) + template + void operator()(WorkQueueBase* self, CALLABLE&& callable, ARGS&&... args) { LLCoros::Promise promise; self->post( - time, // &promise is designed for cross-thread access [&promise, callable = std::move(callable)]() mutable { @@ -558,7 +628,9 @@ namespace LL { promise.set_exception(std::current_exception()); } - }); + }, + // if caller passed a TimePoint, pass it to post() + std::forward(args)...); auto future{ LLCoros::getFuture(promise) }; // block until set_value() LLCoros::TempStatus st("waiting for void WorkQueue::waitForResult()"); @@ -566,13 +638,13 @@ namespace LL } }; - template - auto WorkQueue::waitForResult(const TimePoint& time, CALLABLE&& callable) + template + auto WorkQueueBase::waitForResult(CALLABLE&& callable, ARGS&&... args) { checkCoroutine("waitForResult()"); // derive callable's return type so we can specialize for void return WaitForResult(callable)())>() - (this, time, std::forward(callable)); + (this, std::forward(callable), std::forward(args)...); } } // namespace LL -- cgit v1.2.3 From 424d3ef83cdb354e66789f22f65394f4db523128 Mon Sep 17 00:00:00 2001 From: Nat Goodspeed Date: Tue, 13 Dec 2022 20:49:01 -0500 Subject: DRTVWR-559: Fix broken workqueue_test.cpp. Apparently Visual Studio and Xcode disagree on the intended lifespan of a certain temporary expression. Capturing it in a named variable works. --- indra/llcommon/workqueue.h | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) (limited to 'indra/llcommon/workqueue.h') diff --git a/indra/llcommon/workqueue.h b/indra/llcommon/workqueue.h index eea8886a7a..5461ce6c23 100644 --- a/indra/llcommon/workqueue.h +++ b/indra/llcommon/workqueue.h @@ -419,8 +419,8 @@ namespace LL // moved-from. try { - std::dynamic_pointer_cast(mTarget.lock())-> - post(std::move(*this), mStart); + auto target{ std::dynamic_pointer_cast(mTarget.lock()) }; + target->post(std::move(*this), mStart); } catch (const Closed&) { -- cgit v1.2.3 From 855cae27ccde4896509e3e22c68c441d6404ccfb Mon Sep 17 00:00:00 2001 From: Rye Mutt Date: Fri, 5 May 2023 17:23:29 -0400 Subject: Fix LLThreadSafeQueueInterrupt in WorkQueueBase::postTo during shutdown by catching and returning false --- indra/llcommon/workqueue.h | 68 ++++++++++++++++++++++++++-------------------- 1 file changed, 38 insertions(+), 30 deletions(-) (limited to 'indra/llcommon/workqueue.h') diff --git a/indra/llcommon/workqueue.h b/indra/llcommon/workqueue.h index 5461ce6c23..5b704e7198 100644 --- a/indra/llcommon/workqueue.h +++ b/indra/llcommon/workqueue.h @@ -516,37 +516,45 @@ namespace LL // Here we believe target WorkQueue still exists. Post to it a // lambda that packages our callable, our callback and a weak_ptr // to this originating WorkQueue. - tptr->post( - [reply = super::getWeak(), - callable = std::move(callable), - callback = std::move(callback)] - () mutable - { - // Use postMaybe() below in case this originating WorkQueue - // has been closed or destroyed. Remember, the outer lambda is - // now running on a thread servicing the target WorkQueue, and - // real time has elapsed since postTo()'s tptr->post() call. - try - { - // Make a reply lambda to repost to THIS WorkQueue. - // Delegate to makeReplyLambda() so we can partially - // specialize on void return. - postMaybe(reply, makeReplyLambda(std::move(callable), std::move(callback))); - } - catch (...) + try + { + tptr->post( + [reply = super::getWeak(), + callable = std::move(callable), + callback = std::move(callback)] + () mutable { - // Either variant of makeReplyLambda() is responsible for - // calling the caller's callable. If that throws, return - // the exception to the originating thread. - postMaybe( - reply, - // Bind the current exception to transport back to the - // originating WorkQueue. Once there, rethrow it. - [exc = std::current_exception()](){ std::rethrow_exception(exc); }); - } - }, - // if caller passed a TimePoint, pass it along to post() - std::forward(args)...); + // Use postMaybe() below in case this originating WorkQueue + // has been closed or destroyed. Remember, the outer lambda is + // now running on a thread servicing the target WorkQueue, and + // real time has elapsed since postTo()'s tptr->post() call. + try + { + // Make a reply lambda to repost to THIS WorkQueue. + // Delegate to makeReplyLambda() so we can partially + // specialize on void return. + postMaybe(reply, makeReplyLambda(std::move(callable), std::move(callback))); + } + catch (...) + { + // Either variant of makeReplyLambda() is responsible for + // calling the caller's callable. If that throws, return + // the exception to the originating thread. + postMaybe( + reply, + // Bind the current exception to transport back to the + // originating WorkQueue. Once there, rethrow it. + [exc = std::current_exception()](){ std::rethrow_exception(exc); }); + } + }, + // if caller passed a TimePoint, pass it along to post() + std::forward(args)...); + } + catch (const Closed&) + { + // target WorkQueue still exists, but is Closed + return false; + } // looks like we were able to post() return true; -- cgit v1.2.3 From 026ef1935dbdb21ab79159d38fb78e126dd6ac95 Mon Sep 17 00:00:00 2001 From: Nat Goodspeed Date: Mon, 8 May 2023 12:07:31 -0400 Subject: SL-19690: Follow up on Rye Mutt's fix for shutdown crashes. Rather than continuing to propagate try/catch (Closed) (aka LLThreadSafeQueueInterrupt) constructs through the code base, make WorkQueueBase::post() return bool indicating success (i.e. ! isClosed()). This obviates postIfOpen(), which no one was using anyway. In effect, postIfOpen() is renamed post(), bypassing the exception when isClosed(). Review existing try/catch blocks of that sort, changing to test for post() returning false. --- indra/llcommon/workqueue.h | 135 ++++++++++++++++++--------------------------- 1 file changed, 53 insertions(+), 82 deletions(-) (limited to 'indra/llcommon/workqueue.h') diff --git a/indra/llcommon/workqueue.h b/indra/llcommon/workqueue.h index 5b704e7198..87fdd1517f 100644 --- a/indra/llcommon/workqueue.h +++ b/indra/llcommon/workqueue.h @@ -83,13 +83,10 @@ namespace LL /*---------------------- fire and forget API -----------------------*/ - /// fire-and-forget - virtual void post(const Work&) = 0; - /** * post work, unless the queue is closed before we can post */ - virtual bool postIfOpen(const Work&) = 0; + virtual bool post(const Work&) = 0; /** * post work, unless the queue is full @@ -247,13 +244,10 @@ namespace LL /*---------------------- fire and forget API -----------------------*/ - /// fire-and-forget - void post(const Work&) override; - /** * post work, unless the queue is closed before we can post */ - bool postIfOpen(const Work&) override; + bool post(const Work&) override; /** * post work, unless the queue is full @@ -320,22 +314,16 @@ namespace LL /*---------------------- fire and forget API -----------------------*/ - /// fire-and-forget - void post(const Work& callable) override; - - /// fire-and-forget, but at a particular (future?) time - void post(const Work& callable, const TimePoint& time); - /** * post work, unless the queue is closed before we can post */ - bool postIfOpen(const Work& callable) override; + bool post(const Work& callable) override; /** * post work for a particular time, unless the queue is closed before * we can post */ - bool postIfOpen(const Work& callable, const TimePoint& time); + bool post(const Work& callable, const TimePoint& time); /** * post work, unless the queue is full @@ -356,7 +344,7 @@ namespace LL * an LLCond variant, e.g. LLOneShotCond or LLBoolCond. */ template - void postEvery(const std::chrono::duration& interval, + bool postEvery(const std::chrono::duration& interval, CALLABLE&& callable); private: @@ -417,15 +405,10 @@ namespace LL // move-only callable; but naturally this statement must be // the last time we reference this instance, which may become // moved-from. - try - { - auto target{ std::dynamic_pointer_cast(mTarget.lock()) }; - target->post(std::move(*this), mStart); - } - catch (const Closed&) - { - // Once this queue is closed, oh well, just stop - } + auto target{ std::dynamic_pointer_cast(mTarget.lock()) }; + // Discard bool return: once this queue is closed, oh well, + // just stop + target->post(std::move(*this), mStart); } } @@ -437,7 +420,7 @@ namespace LL }; template - void WorkSchedule::postEvery(const std::chrono::duration& interval, + bool WorkSchedule::postEvery(const std::chrono::duration& interval, CALLABLE&& callable) { if (interval.count() <= 0) @@ -454,7 +437,7 @@ namespace LL // Instantiate and post a suitable BackJack, binding a weak_ptr to // self, the current time, the desired interval and the desired // callable. - post( + return post( BackJack( getWeak(), TimePoint::clock::now(), interval, std::move(callable))); } @@ -516,48 +499,37 @@ namespace LL // Here we believe target WorkQueue still exists. Post to it a // lambda that packages our callable, our callback and a weak_ptr // to this originating WorkQueue. - try - { - tptr->post( - [reply = super::getWeak(), - callable = std::move(callable), - callback = std::move(callback)] - () mutable + return tptr->post( + [reply = super::getWeak(), + callable = std::move(callable), + callback = std::move(callback)] + () mutable + { + // Use postMaybe() below in case this originating WorkQueue + // has been closed or destroyed. Remember, the outer lambda is + // now running on a thread servicing the target WorkQueue, and + // real time has elapsed since postTo()'s tptr->post() call. + try { - // Use postMaybe() below in case this originating WorkQueue - // has been closed or destroyed. Remember, the outer lambda is - // now running on a thread servicing the target WorkQueue, and - // real time has elapsed since postTo()'s tptr->post() call. - try - { - // Make a reply lambda to repost to THIS WorkQueue. - // Delegate to makeReplyLambda() so we can partially - // specialize on void return. - postMaybe(reply, makeReplyLambda(std::move(callable), std::move(callback))); - } - catch (...) - { - // Either variant of makeReplyLambda() is responsible for - // calling the caller's callable. If that throws, return - // the exception to the originating thread. - postMaybe( - reply, - // Bind the current exception to transport back to the - // originating WorkQueue. Once there, rethrow it. - [exc = std::current_exception()](){ std::rethrow_exception(exc); }); - } - }, - // if caller passed a TimePoint, pass it along to post() - std::forward(args)...); - } - catch (const Closed&) - { - // target WorkQueue still exists, but is Closed - return false; - } - - // looks like we were able to post() - return true; + // Make a reply lambda to repost to THIS WorkQueue. + // Delegate to makeReplyLambda() so we can partially + // specialize on void return. + postMaybe(reply, makeReplyLambda(std::move(callable), std::move(callback))); + } + catch (...) + { + // Either variant of makeReplyLambda() is responsible for + // calling the caller's callable. If that throws, return + // the exception to the originating thread. + postMaybe( + reply, + // Bind the current exception to transport back to the + // originating WorkQueue. Once there, rethrow it. + [exc = std::current_exception()](){ std::rethrow_exception(exc); }); + } + }, + // if caller passed a TimePoint, pass it along to post() + std::forward(args)...); } template @@ -568,18 +540,9 @@ namespace LL auto tptr = target.lock(); if (tptr) { - try - { - tptr->post(std::forward(args)...); - // we were able to post() - return true; - } - catch (const Closed&) - { - // target WorkQueue still exists, but is Closed - } + return tptr->post(std::forward(args)...); } - // either target no longer exists, or its WorkQueue is Closed + // target no longer exists return false; } @@ -591,7 +554,7 @@ namespace LL auto operator()(WorkQueueBase* self, CALLABLE&& callable, ARGS&&... args) { LLCoros::Promise promise; - self->post( + bool posted = self->post( // We dare to bind a reference to Promise because it's // specifically designed for cross-thread communication. [&promise, callable = std::move(callable)]() @@ -608,6 +571,10 @@ namespace LL }, // if caller passed a TimePoint, pass it to post() std::forward(args)...); + if (! posted) + { + LLTHROW(Closed); + } auto future{ LLCoros::getFuture(promise) }; // now, on the calling thread, wait for that result LLCoros::TempStatus st("waiting for WorkQueue::waitForResult()"); @@ -623,7 +590,7 @@ namespace LL void operator()(WorkQueueBase* self, CALLABLE&& callable, ARGS&&... args) { LLCoros::Promise promise; - self->post( + bool posted = self->post( // &promise is designed for cross-thread access [&promise, callable = std::move(callable)]() mutable { @@ -639,6 +606,10 @@ namespace LL }, // if caller passed a TimePoint, pass it to post() std::forward(args)...); + if (! posted) + { + LLTHROW(Closed); + } auto future{ LLCoros::getFuture(promise) }; // block until set_value() LLCoros::TempStatus st("waiting for void WorkQueue::waitForResult()"); -- cgit v1.2.3 From d75ecde69c1289f4e3df8b75e9a74c5b05db318c Mon Sep 17 00:00:00 2001 From: Nat Goodspeed Date: Mon, 8 May 2023 12:31:57 -0400 Subject: SL-19690: Properly qualify exception type. --- indra/llcommon/workqueue.h | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) (limited to 'indra/llcommon/workqueue.h') diff --git a/indra/llcommon/workqueue.h b/indra/llcommon/workqueue.h index 87fdd1517f..ec0700a718 100644 --- a/indra/llcommon/workqueue.h +++ b/indra/llcommon/workqueue.h @@ -573,7 +573,7 @@ namespace LL std::forward(args)...); if (! posted) { - LLTHROW(Closed); + LLTHROW(WorkQueueBase::Closed()); } auto future{ LLCoros::getFuture(promise) }; // now, on the calling thread, wait for that result @@ -608,7 +608,7 @@ namespace LL std::forward(args)...); if (! posted) { - LLTHROW(Closed); + LLTHROW(WorkQueueBase::Closed()); } auto future{ LLCoros::getFuture(promise) }; // block until set_value() -- cgit v1.2.3