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 <mutex>
17 #include <thread>
18
19 #include <base/containers/vector.h>
20
21 #include <meta/interface/intf_task_queue.h>
22 #include <meta/interface/intf_task_queue_registry.h>
23
24 #include "future.h"
25 #include "object.h"
26 #include "task_queue.h"
27
28 META_BEGIN_NAMESPACE()
29
30 // notice, this is object only so we can construct it via object registery
31 class PollingTaskQueue : public IntroduceInterfaces<MetaObject, IPollingTaskQueue, TaskQueueImpl> {
32 META_OBJECT(PollingTaskQueue, ClassId::PollingTaskQueue, IntroduceInterfaces)
33 public:
34 using Token = ITaskQueue::Token;
35
Build(const IMetadata::Ptr & data)36 bool Build(const IMetadata::Ptr& data) override
37 {
38 bool ret = Super::Build(data);
39 if (ret) {
40 self_ = GetSelf<ITaskQueue>();
41 }
42 return ret;
43 }
44
InvokeTask(const ITaskQueueTask::Ptr & task)45 bool InvokeTask(const ITaskQueueTask::Ptr& task) override
46 {
47 auto q = GetTaskQueueRegistry().SetCurrentTaskQueue(self_);
48 auto ret = task->Invoke();
49 GetTaskQueueRegistry().SetCurrentTaskQueue(q);
50 return ret;
51 }
CancelTask(Token token)52 void CancelTask(Token token) override
53 {
54 TaskQueueImpl::CancelTask(token);
55 }
56
AddTask(ITaskQueueTask::Ptr p)57 Token AddTask(ITaskQueueTask::Ptr p) override
58 {
59 return AddTask(BASE_NS::move(p), TimeSpan::Milliseconds(0));
60 }
61
AddTask(ITaskQueueTask::Ptr p,const TimeSpan & delay)62 Token AddTask(ITaskQueueTask::Ptr p, const TimeSpan& delay) override
63 {
64 return TaskQueueImpl::AddTask(BASE_NS::move(p), delay, Time() + delay);
65 }
66
AddWaitableTask(ITaskQueueWaitableTask::Ptr p)67 IFuture::Ptr AddWaitableTask(ITaskQueueWaitableTask::Ptr p) override
68 {
69 IPromise::Ptr promise(new Promise);
70 BASE_NS::shared_ptr<PromisedQueueTask> task(new PromisedQueueTask(BASE_NS::move(p), promise));
71 auto f = task->GetFuture();
72 AddTask(BASE_NS::move(task));
73 return f;
74 }
75
ProcessTasks()76 void ProcessTasks() override
77 {
78 TimeSpan ctime = Time();
79 std::unique_lock lock { mutex_ };
80 if (ctime != lastTime_) {
81 lastTime_ = ctime;
82 execThread_ = std::this_thread::get_id();
83 TaskQueueImpl::ProcessTasks(lock, ctime);
84 execThread_ = std::thread::id();
85 } else {
86 // non issue. but.
87 CORE_LOG_V("Double call to ProcessTasks.");
88 }
89 }
90
91 private:
92 TimeSpan lastTime_;
93 };
94 // Internal api for engine task queue
95 namespace Internal {
96
GetPollingTaskQueueFactory()97 IObjectFactory::Ptr GetPollingTaskQueueFactory()
98 {
99 return PollingTaskQueue::GetFactory();
100 }
101
102 } // namespace Internal
103
104 META_END_NAMESPACE()
105