1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
|
/*
* 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/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 MessageHandlerAutoCleanup {
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);
};
} // namespace rtc
#endif // RTC_BASE_ASYNC_INVOKER_H_
|
