1 /*
2 * Copyright (c) 2024 Huawei Device Co., Ltd.
3 * Licensed under the Apache License, Version 2.0 (the "License");
4 * you may not use this file except in compliance with the License.
5 * You may obtain a copy of the License at
6 *
7 * http://www.apache.org/licenses/LICENSE-2.0
8 *
9 * Unless required by applicable law or agreed to in writing, software
10 * distributed under the License is distributed on an "AS IS" BASIS,
11 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 * See the License for the specific language governing permissions and
13 * limitations under the License.
14 */
15
16 #include "future.h"
17
18 #include <meta/interface/object_type_info.h>
19
META_BEGIN_NAMESPACE()20 META_BEGIN_NAMESPACE()
21
22 Future::StateType Future::GetState() const
23 {
24 std::unique_lock lock { mutex_ };
25 return state_;
26 }
27
Wait() const28 Future::StateType Future::Wait() const
29 {
30 std::unique_lock lock { mutex_ };
31 while (state_ == IFuture::WAITING) {
32 cond_.wait(lock);
33 }
34 return state_;
35 }
36
WaitFor(const TimeSpan & time) const37 Future::StateType Future::WaitFor(const TimeSpan& time) const
38 {
39 std::unique_lock lock { mutex_ };
40 while (state_ == IFuture::WAITING) {
41 if (cond_.wait_for(lock, std::chrono::microseconds(time.ToMicroseconds())) == std::cv_status::timeout) {
42 return IFuture::WAITING;
43 }
44 }
45 return state_;
46 }
47
GetResult() const48 IAny::Ptr Future::GetResult() const
49 {
50 std::unique_lock lock { mutex_ };
51 while (state_ == IFuture::WAITING) {
52 cond_.wait(lock);
53 }
54 return result_;
55 }
56
Then(const IFutureContinuation::Ptr & func,const ITaskQueue::Ptr & queue)57 IFuture::Ptr Future::Then(const IFutureContinuation::Ptr& func, const ITaskQueue::Ptr& queue)
58 {
59 std::unique_lock lock { mutex_ };
60 IFuture::Ptr result;
61 if (state_ == IFuture::ABANDONED) {
62 BASE_NS::shared_ptr<Future> f(new Future);
63 f->SetAbandoned();
64 result = BASE_NS::move(f);
65 } else {
66 ContinuationData d { queue == nullptr, queue };
67 d.continuation.reset(new ContinuationQueueTask(func));
68 result = d.continuation->GetFuture();
69 continuations_.push_back(BASE_NS::move(d));
70 if (state_ == IFuture::COMPLETED) {
71 ActivateContinuation(lock);
72 }
73 }
74 return result;
75 }
76
Cancel()77 void Future::Cancel()
78 {
79 bool notify = false;
80 ITaskQueue::Token token {};
81 {
82 std::unique_lock lock { mutex_ };
83 if (state_ != IFuture::COMPLETED) {
84 notify = true;
85 token = token_;
86 token_ = {};
87 for (auto&& v : continuations_) {
88 v.continuation->SetAbandoned();
89 }
90 continuations_.clear();
91 state_ = IFuture::ABANDONED;
92 }
93 }
94 if (token) {
95 if (auto q = queue_.lock()) {
96 q->CancelTask(token);
97 }
98 }
99 if (notify) {
100 cond_.notify_all();
101 }
102 }
103
ActivateContinuation(const ContinuationData & d,const IAny::Ptr & result)104 void Future::ActivateContinuation(const ContinuationData& d, const IAny::Ptr& result)
105 {
106 if (auto q = d.queue.lock()) {
107 d.continuation->SetParam(result);
108 auto token = q->AddTask(d.continuation);
109 d.continuation->SetQueueInfo(q, token);
110 } else if (d.runInline) {
111 d.continuation->SetParam(result);
112 d.continuation->Invoke();
113 } else {
114 d.continuation->SetAbandoned();
115 }
116 }
117
ActivateContinuation(std::unique_lock<std::mutex> & lock)118 void Future::ActivateContinuation(std::unique_lock<std::mutex>& lock)
119 {
120 BASE_NS::vector<ContinuationData> cdata = BASE_NS::move(continuations_);
121 auto result = result_;
122 lock.unlock();
123 for (auto&& v : cdata) {
124 ActivateContinuation(v, result);
125 }
126 }
127
SetResult(IAny::Ptr p)128 void Future::SetResult(IAny::Ptr p)
129 {
130 std::unique_lock lock { mutex_ };
131 token_ = {};
132 if (state_ == IFuture::WAITING) {
133 result_ = BASE_NS::move(p);
134 state_ = IFuture::COMPLETED;
135 ActivateContinuation(lock);
136 cond_.notify_all();
137 }
138 }
139
SetAbandoned()140 void Future::SetAbandoned()
141 {
142 std::unique_lock lock { mutex_ };
143 if (state_ == IFuture::WAITING) {
144 state_ = IFuture::ABANDONED;
145 token_ = {};
146
147 for (auto&& v : continuations_) {
148 v.continuation->SetAbandoned();
149 }
150 continuations_.clear();
151 cond_.notify_all();
152 }
153 }
154
SetQueueInfo(const ITaskQueue::Ptr & queue,ITaskQueue::Token token)155 void Future::SetQueueInfo(const ITaskQueue::Ptr& queue, ITaskQueue::Token token)
156 {
157 std::unique_lock lock { mutex_ };
158 if (state_ == IFuture::WAITING) {
159 queue_ = queue;
160 token_ = token;
161 }
162 }
163
~Promise()164 Promise::~Promise()
165 {
166 SetAbandoned();
167 }
168
Set(const IAny::Ptr & res)169 void Promise::Set(const IAny::Ptr& res)
170 {
171 if (future_) {
172 future_->SetResult(res);
173 future_ = nullptr;
174 }
175 }
176
SetAbandoned()177 void Promise::SetAbandoned()
178 {
179 if (future_) {
180 future_->SetAbandoned();
181 future_ = nullptr;
182 }
183 }
184
GetFuture()185 IFuture::Ptr Promise::GetFuture()
186 {
187 if (!future_) {
188 future_.reset(new Future);
189 }
190 return future_;
191 }
192
SetQueueInfo(const ITaskQueue::Ptr & queue,ITaskQueue::Token token)193 void Promise::SetQueueInfo(const ITaskQueue::Ptr& queue, ITaskQueue::Token token)
194 {
195 if (future_) {
196 future_->SetQueueInfo(queue, token);
197 }
198 }
199
200 namespace Internal {
201
GetPromiseFactory()202 IObjectFactory::Ptr GetPromiseFactory()
203 {
204 return Promise::GetFactory();
205 }
206
207 } // namespace Internal
208
209 META_END_NAMESPACE()
210