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 "threading/task_queue.h"
17
18 #include <atomic>
19
20 #include <base/containers/refcnt_ptr.h>
21 #include <base/containers/type_traits.h>
22 #include <base/containers/unique_ptr.h>
23 #include <core/log.h>
24 #include <core/namespace.h>
25 #include <core/threading/intf_thread_pool.h>
26
27 CORE_BEGIN_NAMESPACE()
28 using BASE_NS::move;
29 using BASE_NS::unique_ptr;
30
31 // -- TaskQueue ExecuteAsyncTask, runs TaskQueue::Execute.
ExecuteAsyncTask(TaskQueue & queue)32 TaskQueue::ExecuteAsyncTask::ExecuteAsyncTask(TaskQueue& queue) : queue_(queue) {}
33
operator ()()34 void TaskQueue::ExecuteAsyncTask::operator()()
35 {
36 queue_.Execute();
37 queue_.isRunningAsync_ = false;
38 }
39
Destroy()40 void TaskQueue::ExecuteAsyncTask::Destroy()
41 {
42 delete this;
43 }
44
45 // -- TaskQueue
TaskQueue(const IThreadPool::Ptr & threadPool)46 TaskQueue::TaskQueue(const IThreadPool::Ptr& threadPool) : threadPool_(threadPool), isRunningAsync_(false) {}
47
48 TaskQueue::~TaskQueue() = default;
49
ExecuteAsync()50 void TaskQueue::ExecuteAsync()
51 {
52 CORE_ASSERT(threadPool_ != nullptr);
53
54 if (!IsRunningAsync()) {
55 isRunningAsync_ = true;
56
57 // Execute in new thread.
58 asyncOperation_ = threadPool_->Push(IThreadPool::ITask::Ptr { new ExecuteAsyncTask(*this) });
59 }
60 }
61
IsRunningAsync() const62 bool TaskQueue::IsRunningAsync() const
63 {
64 return isRunningAsync_;
65 }
66
Wait()67 void TaskQueue::Wait()
68 {
69 if (IsRunningAsync()) {
70 asyncOperation_->Wait();
71 isRunningAsync_ = false;
72 }
73 }
74
75 // -- TaskQueue entry.
Entry(uint64_t identifier,IThreadPool::ITask::Ptr task)76 TaskQueue::Entry::Entry(uint64_t identifier, IThreadPool::ITask::Ptr task) : task(move(task)), identifier(identifier) {}
77
operator ==(uint64_t rhsIdentifier) const78 bool TaskQueue::Entry::operator==(uint64_t rhsIdentifier) const
79 {
80 return identifier == rhsIdentifier;
81 }
82
operator ==(const TaskQueue::Entry & other) const83 bool TaskQueue::Entry::operator==(const TaskQueue::Entry& other) const
84 {
85 return identifier == other.identifier;
86 }
87 CORE_END_NAMESPACE()
88