123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265 |
- /*
- * Copyright 2014 The WebRTC Project Authors. All rights reserved.
- *
- * Use of this source code is governed by a BSD-style license
- * that can be found in the LICENSE file in the root of the source
- * tree. An additional intellectual property rights grant can be found
- * in the file PATENTS. All contributing project authors may
- * be found in the AUTHORS file in the root of the source tree.
- */
- #ifndef RTC_BASE_ASYNC_INVOKER_H_
- #define RTC_BASE_ASYNC_INVOKER_H_
- #include <atomic>
- #include <memory>
- #include <utility>
- #include "api/scoped_refptr.h"
- #include "rtc_base/async_invoker_inl.h"
- #include "rtc_base/bind.h"
- #include "rtc_base/constructor_magic.h"
- #include "rtc_base/event.h"
- #include "rtc_base/ref_counted_object.h"
- #include "rtc_base/third_party/sigslot/sigslot.h"
- #include "rtc_base/thread.h"
- namespace rtc {
- // Invokes function objects (aka functors) asynchronously on a Thread, and
- // owns the lifetime of calls (ie, when this object is destroyed, calls in
- // flight are cancelled). AsyncInvoker can optionally execute a user-specified
- // function when the asynchronous call is complete, or operates in
- // fire-and-forget mode otherwise.
- //
- // AsyncInvoker does not own the thread it calls functors on.
- //
- // A note about async calls and object lifetimes: users should
- // be mindful of object lifetimes when calling functions asynchronously and
- // ensure objects used by the function _cannot_ be deleted between the
- // invocation and execution of the functor. AsyncInvoker is designed to
- // help: any calls in flight will be cancelled when the AsyncInvoker used to
- // make the call is destructed, and any calls executing will be allowed to
- // complete before AsyncInvoker destructs.
- //
- // The easiest way to ensure lifetimes are handled correctly is to create a
- // class that owns the Thread and AsyncInvoker objects, and then call its
- // methods asynchronously as needed.
- //
- // Example:
- // class MyClass {
- // public:
- // void FireAsyncTaskWithResult(Thread* thread, int x) {
- // // Specify a callback to get the result upon completion.
- // invoker_.AsyncInvoke<int>(RTC_FROM_HERE,
- // thread, Bind(&MyClass::AsyncTaskWithResult, this, x),
- // &MyClass::OnTaskComplete, this);
- // }
- // void FireAnotherAsyncTask(Thread* thread) {
- // // No callback specified means fire-and-forget.
- // invoker_.AsyncInvoke<void>(RTC_FROM_HERE,
- // thread, Bind(&MyClass::AnotherAsyncTask, this));
- //
- // private:
- // int AsyncTaskWithResult(int x) {
- // // Some long running process...
- // return x * x;
- // }
- // void AnotherAsyncTask() {
- // // Some other long running process...
- // }
- // void OnTaskComplete(int result) { result_ = result; }
- //
- // AsyncInvoker invoker_;
- // int result_;
- // };
- //
- // More details about threading:
- // - It's safe to construct/destruct AsyncInvoker on different threads.
- // - It's safe to call AsyncInvoke from different threads.
- // - It's safe to call AsyncInvoke recursively from *within* a functor that's
- // being AsyncInvoked.
- // - However, it's *not* safe to call AsyncInvoke from *outside* a functor
- // that's being AsyncInvoked while the AsyncInvoker is being destroyed on
- // another thread. This is just inherently unsafe and there's no way to
- // prevent that. So, the user of this class should ensure that the start of
- // each "chain" of invocations is synchronized somehow with the AsyncInvoker's
- // destruction. This can be done by starting each chain of invocations on the
- // same thread on which it will be destroyed, or by using some other
- // synchronization method.
- class AsyncInvoker : public MessageHandler {
- public:
- AsyncInvoker();
- ~AsyncInvoker() override;
- // Call |functor| asynchronously on |thread|, with no callback upon
- // completion. Returns immediately.
- template <class ReturnT, class FunctorT>
- void AsyncInvoke(const Location& posted_from,
- Thread* thread,
- FunctorT&& functor,
- uint32_t id = 0) {
- std::unique_ptr<AsyncClosure> closure(
- new FireAndForgetAsyncClosure<FunctorT>(
- this, std::forward<FunctorT>(functor)));
- DoInvoke(posted_from, thread, std::move(closure), id);
- }
- // Call |functor| asynchronously on |thread| with |delay_ms|, with no callback
- // upon completion. Returns immediately.
- template <class ReturnT, class FunctorT>
- void AsyncInvokeDelayed(const Location& posted_from,
- Thread* thread,
- FunctorT&& functor,
- uint32_t delay_ms,
- uint32_t id = 0) {
- std::unique_ptr<AsyncClosure> closure(
- new FireAndForgetAsyncClosure<FunctorT>(
- this, std::forward<FunctorT>(functor)));
- DoInvokeDelayed(posted_from, thread, std::move(closure), delay_ms, id);
- }
- // Synchronously execute on |thread| all outstanding calls we own
- // that are pending on |thread|, and wait for calls to complete
- // before returning. Optionally filter by message id.
- // The destructor will not wait for outstanding calls, so if that
- // behavior is desired, call Flush() before destroying this object.
- void Flush(Thread* thread, uint32_t id = MQID_ANY);
- // Cancels any outstanding calls we own that are pending on any thread, and
- // which have not yet started to execute. This does not wait for any calls
- // that have already started executing to complete.
- void Clear();
- private:
- void OnMessage(Message* msg) override;
- void DoInvoke(const Location& posted_from,
- Thread* thread,
- std::unique_ptr<AsyncClosure> closure,
- uint32_t id);
- void DoInvokeDelayed(const Location& posted_from,
- Thread* thread,
- std::unique_ptr<AsyncClosure> closure,
- uint32_t delay_ms,
- uint32_t id);
- // Used to keep track of how many invocations (AsyncClosures) are still
- // alive, so that the destructor can wait for them to finish, as described in
- // the class documentation.
- //
- // TODO(deadbeef): Using a raw std::atomic like this is prone to error and
- // difficult to maintain. We should try to wrap this functionality in a
- // separate class to reduce the chance of errors being introduced in the
- // future.
- std::atomic<int> pending_invocations_;
- // Reference counted so that if the AsyncInvoker destructor finishes before
- // an AsyncClosure's destructor that's about to call
- // "invocation_complete_->Set()", it's not dereferenced after being
- // destroyed.
- scoped_refptr<RefCountedObject<Event>> invocation_complete_;
- // This flag is used to ensure that if an application AsyncInvokes tasks that
- // recursively AsyncInvoke other tasks ad infinitum, the cycle eventually
- // terminates.
- std::atomic<bool> destroying_;
- friend class AsyncClosure;
- RTC_DISALLOW_COPY_AND_ASSIGN(AsyncInvoker);
- };
- // Similar to AsyncInvoker, but guards against the Thread being destroyed while
- // there are outstanding dangling pointers to it. It will connect to the current
- // thread in the constructor, and will get notified when that thread is
- // destroyed. After GuardedAsyncInvoker is constructed, it can be used from
- // other threads to post functors to the thread it was constructed on. If that
- // thread dies, any further calls to AsyncInvoke() will be safely ignored.
- class GuardedAsyncInvoker : public sigslot::has_slots<> {
- public:
- GuardedAsyncInvoker();
- ~GuardedAsyncInvoker() override;
- // Synchronously execute all outstanding calls we own, and wait for calls to
- // complete before returning. Optionally filter by message id. The destructor
- // will not wait for outstanding calls, so if that behavior is desired, call
- // Flush() first. Returns false if the thread has died.
- bool Flush(uint32_t id = MQID_ANY);
- // Call |functor| asynchronously with no callback upon completion. Returns
- // immediately. Returns false if the thread has died.
- template <class ReturnT, class FunctorT>
- bool AsyncInvoke(const Location& posted_from,
- FunctorT&& functor,
- uint32_t id = 0) {
- CritScope cs(&crit_);
- if (thread_ == nullptr)
- return false;
- invoker_.AsyncInvoke<ReturnT, FunctorT>(
- posted_from, thread_, std::forward<FunctorT>(functor), id);
- return true;
- }
- // Call |functor| asynchronously with |delay_ms|, with no callback upon
- // completion. Returns immediately. Returns false if the thread has died.
- template <class ReturnT, class FunctorT>
- bool AsyncInvokeDelayed(const Location& posted_from,
- FunctorT&& functor,
- uint32_t delay_ms,
- uint32_t id = 0) {
- CritScope cs(&crit_);
- if (thread_ == nullptr)
- return false;
- invoker_.AsyncInvokeDelayed<ReturnT, FunctorT>(
- posted_from, thread_, std::forward<FunctorT>(functor), delay_ms, id);
- return true;
- }
- // Call |functor| asynchronously, calling |callback| when done. Returns false
- // if the thread has died.
- template <class ReturnT, class FunctorT, class HostT>
- bool AsyncInvoke(const Location& posted_from,
- const Location& callback_posted_from,
- FunctorT&& functor,
- void (HostT::*callback)(ReturnT),
- HostT* callback_host,
- uint32_t id = 0) {
- CritScope cs(&crit_);
- if (thread_ == nullptr)
- return false;
- invoker_.AsyncInvoke<ReturnT, FunctorT, HostT>(
- posted_from, callback_posted_from, thread_,
- std::forward<FunctorT>(functor), callback, callback_host, id);
- return true;
- }
- // Call |functor| asynchronously calling |callback| when done. Overloaded for
- // void return. Returns false if the thread has died.
- template <class ReturnT, class FunctorT, class HostT>
- bool AsyncInvoke(const Location& posted_from,
- const Location& callback_posted_from,
- FunctorT&& functor,
- void (HostT::*callback)(),
- HostT* callback_host,
- uint32_t id = 0) {
- CritScope cs(&crit_);
- if (thread_ == nullptr)
- return false;
- invoker_.AsyncInvoke<ReturnT, FunctorT, HostT>(
- posted_from, callback_posted_from, thread_,
- std::forward<FunctorT>(functor), callback, callback_host, id);
- return true;
- }
- private:
- // Callback when |thread_| is destroyed.
- void ThreadDestroyed();
- CriticalSection crit_;
- Thread* thread_ RTC_GUARDED_BY(crit_);
- AsyncInvoker invoker_ RTC_GUARDED_BY(crit_);
- };
- } // namespace rtc
- #endif // RTC_BASE_ASYNC_INVOKER_H_
|