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 "meta_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
32 : public Internal::MetaObjectFwd<PollingTaskQueue, ClassId::PollingTaskQueue, IPollingTaskQueue, TaskQueueImpl> {
33 public:
34 using Super =
35 Internal::MetaObjectFwd<PollingTaskQueue, ClassId::PollingTaskQueue, IPollingTaskQueue, TaskQueueImpl>;
36 using Token = ITaskQueue::Token;
37
Build(const IMetadata::Ptr & data)38 bool Build(const IMetadata::Ptr& data) override
39 {
40 bool ret = Super::Build(data);
41 if (ret) {
42 self_ = GetSelf<ITaskQueue>();
43 }
44 return ret;
45 }
46
InvokeTask(const ITaskQueueTask::Ptr & task)47 bool InvokeTask(const ITaskQueueTask::Ptr& task) override
48 {
49 auto q = GetTaskQueueRegistry().SetCurrentTaskQueue(self_);
50 auto ret = task->Invoke();
51 GetTaskQueueRegistry().SetCurrentTaskQueue(q);
52 return ret;
53 }
CancelTask(Token token)54 void CancelTask(Token token) override
55 {
56 TaskQueueImpl::CancelTask(token);
57 }
58
AddTask(ITaskQueueTask::Ptr p)59 Token AddTask(ITaskQueueTask::Ptr p) override
60 {
61 return AddTask(BASE_NS::move(p), TimeSpan::Milliseconds(0));
62 }
63
AddTask(ITaskQueueTask::Ptr p,const TimeSpan & delay)64 Token AddTask(ITaskQueueTask::Ptr p, const TimeSpan& delay) override
65 {
66 return TaskQueueImpl::AddTask(BASE_NS::move(p), delay, Time() + delay);
67 }
68
AddWaitableTask(ITaskQueueWaitableTask::Ptr p)69 IFuture::Ptr AddWaitableTask(ITaskQueueWaitableTask::Ptr p) override
70 {
71 IPromise::Ptr promise(new Promise);
72 BASE_NS::shared_ptr<PromisedQueueTask> task(new PromisedQueueTask(BASE_NS::move(p), promise));
73 auto f = task->GetFuture();
74 AddTask(BASE_NS::move(task));
75 return f;
76 }
77
ProcessTasks()78 void ProcessTasks() override
79 {
80 TimeSpan ctime = Time();
81 std::unique_lock lock { mutex_ };
82 if (ctime != lastTime_) {
83 lastTime_ = ctime;
84 execThread_ = std::this_thread::get_id();
85 TaskQueueImpl::ProcessTasks(lock, ctime);
86 execThread_ = std::thread::id();
87 } else {
88 // non issue. but.
89 CORE_LOG_V("Double call to ProcessTasks.");
90 }
91 }
92
93 private:
94 TimeSpan lastTime_;
95 };
96 // Internal api for engine task queue
97 namespace Internal {
98
GetPollingTaskQueueFactory()99 IObjectFactory::Ptr GetPollingTaskQueueFactory()
100 {
101 return PollingTaskQueue::GetFactory();
102 }
103
104 } // namespace Internal
105
106 META_END_NAMESPACE()
107