diff options
Diffstat (limited to 'indra/llcommon/workqueue.h')
| -rw-r--r-- | indra/llcommon/workqueue.h | 184 |
1 files changed, 184 insertions, 0 deletions
diff --git a/indra/llcommon/workqueue.h b/indra/llcommon/workqueue.h index 5ec790da79..8e4b38c2f3 100644 --- a/indra/llcommon/workqueue.h +++ b/indra/llcommon/workqueue.h @@ -222,6 +222,7 @@ namespace LL template <typename Rep, typename Period> bool runFor(const std::chrono::duration<Rep, Period>& timeslice) { + LL_PROFILE_ZONE_SCOPED; return runUntil(TimePoint::clock::now() + timeslice); } @@ -330,6 +331,189 @@ namespace LL getWeak(), TimePoint::clock::now(), interval, std::move(callable))); } + /// general case: arbitrary C++ return type + template <typename CALLABLE, typename FOLLOWUP, typename RETURNTYPE> + struct WorkQueue::MakeReplyLambda + { + auto operator()(CALLABLE&& callable, FOLLOWUP&& callback) + { + // Call the callable in any case -- but to minimize + // copying the result, immediately bind it into the reply + // lambda. The reply lambda also binds the original + // callback, so that when we, the originating WorkQueue, + // finally receive and process the reply lambda, we'll + // call the bound callback with the bound result -- on the + // same thread that originally called postTo(). + return + [result = std::forward<CALLABLE>(callable)(), + callback = std::move(callback)] + () + { callback(std::move(result)); }; + } + }; + + /// specialize for CALLABLE returning void + template <typename CALLABLE, typename FOLLOWUP> + struct WorkQueue::MakeReplyLambda<CALLABLE, FOLLOWUP, void> + { + auto operator()(CALLABLE&& callable, FOLLOWUP&& callback) + { + // Call the callable, which produces no result. + std::forward<CALLABLE>(callable)(); + // Our completion callback is simply the caller's callback. + return std::move(callback); + } + }; + + template <typename CALLABLE, typename FOLLOWUP> + auto WorkQueue::makeReplyLambda(CALLABLE&& callable, FOLLOWUP&& callback) + { + return MakeReplyLambda<CALLABLE, FOLLOWUP, + decltype(std::forward<CALLABLE>(callable)())>() + (std::move(callable), std::move(callback)); + } + + template <typename CALLABLE, typename FOLLOWUP> + bool WorkQueue::postTo(weak_t target, + const TimePoint& time, CALLABLE&& callable, FOLLOWUP&& callback) + { + LL_PROFILE_ZONE_SCOPED; + // We're being asked to post to the WorkQueue at target. + // target is a weak_ptr: have to lock it to check it. + auto tptr = target.lock(); + if (! tptr) + // can't post() if the target WorkQueue has been destroyed + return false; + + // 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( + time, + [reply = super::getWeak(), + callable = std::move(callable), + callback = std::move(callback)] + () + { + // 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); }); + } + }); + + // looks like we were able to post() + return true; + } + + template <typename CALLABLE> + bool WorkQueue::postMaybe(weak_t target, const TimePoint& time, CALLABLE&& callable) + { + LL_PROFILE_ZONE_SCOPED; + // target is a weak_ptr: have to lock it to check it + auto tptr = target.lock(); + if (tptr) + { + try + { + tptr->post(time, std::forward<CALLABLE>(callable)); + // we were able to post() + return true; + } + catch (const Closed&) + { + // target WorkQueue still exists, but is Closed + } + } + // either target no longer exists, or its WorkQueue is Closed + return false; + } + + /// general case: arbitrary C++ return type + template <typename CALLABLE, typename RETURNTYPE> + struct WorkQueue::WaitForResult + { + auto operator()(WorkQueue* self, const TimePoint& time, CALLABLE&& callable) + { + LLCoros::Promise<RETURNTYPE> 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)]() + { + try + { + // call the caller's callable and trigger promise with result + promise.set_value(callable()); + } + catch (...) + { + promise.set_exception(std::current_exception()); + } + }); + auto future{ LLCoros::getFuture(promise) }; + // now, on the calling thread, wait for that result + LLCoros::TempStatus st("waiting for WorkQueue::waitForResult()"); + return future.get(); + } + }; + + /// specialize for CALLABLE returning void + template <typename CALLABLE> + struct WorkQueue::WaitForResult<CALLABLE, void> + { + void operator()(WorkQueue* self, const TimePoint& time, CALLABLE&& callable) + { + LLCoros::Promise<void> promise; + self->post( + time, + // &promise is designed for cross-thread access + [&promise, callable = std::move(callable)]() + { + try + { + callable(); + promise.set_value(); + } + catch (...) + { + promise.set_exception(std::current_exception()); + } + }); + auto future{ LLCoros::getFuture(promise) }; + // block until set_value() + LLCoros::TempStatus st("waiting for void WorkQueue::waitForResult()"); + future.get(); + } + }; + + template <typename CALLABLE> + auto WorkQueue::waitForResult(const TimePoint& time, CALLABLE&& callable) + { + checkCoroutine("waitForResult()"); + // derive callable's return type so we can specialize for void + return WaitForResult<CALLABLE, decltype(std::forward<CALLABLE>(callable)())>() + (this, time, std::forward<CALLABLE>(callable)); + } + } // namespace LL #endif /* ! defined(LL_WORKQUEUE_H) */ |