1 /*
2 * Copyright (C) 2023 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 <condition_variable>
19 #include <mutex>
20 #include <queue>
21 #include <thread>
22
23 #include <base/containers/vector.h>
24 #include <core/log.h>
25 #include <core/namespace.h>
26
27 #include "os/platform.h"
28
29 CORE_BEGIN_NAMESPACE()
30 using BASE_NS::array_view;
31 using BASE_NS::make_unique;
32 using BASE_NS::move;
33 using BASE_NS::unique_ptr;
34
35 // -- TaskQueue ExecuteAsyncTask, runs TaskQueue::Execute.
ExecuteAsyncTask(TaskQueue & queue)36 TaskQueue::ExecuteAsyncTask::ExecuteAsyncTask(TaskQueue& queue) : queue_(queue) {}
37
operator ()()38 void TaskQueue::ExecuteAsyncTask::operator()()
39 {
40 queue_.Execute();
41 queue_.isRunningAsync_ = false;
42 }
43
Destroy()44 void TaskQueue::ExecuteAsyncTask::Destroy()
45 {
46 delete this;
47 }
48
49 // -- TaskQueue
TaskQueue(const IThreadPool::Ptr & threadPool)50 TaskQueue::TaskQueue(const IThreadPool::Ptr& threadPool) : threadPool_(threadPool), isRunningAsync_(false) {}
51
52 TaskQueue::~TaskQueue() = default;
53
ExecuteAsync()54 void TaskQueue::ExecuteAsync()
55 {
56 CORE_ASSERT(threadPool_ != nullptr);
57
58 if (!IsRunningAsync()) {
59 isRunningAsync_ = true;
60
61 // Execute in new thread.
62 asyncOperation_ = threadPool_->Push(IThreadPool::ITask::Ptr { new ExecuteAsyncTask(*this) });
63 }
64 }
65
IsRunningAsync() const66 bool TaskQueue::IsRunningAsync() const
67 {
68 return isRunningAsync_;
69 }
70
Wait()71 void TaskQueue::Wait()
72 {
73 if (IsRunningAsync()) {
74 asyncOperation_->Wait();
75 isRunningAsync_ = false;
76 }
77 }
78
79 // -- TaskQueue entry.
Entry()80 TaskQueue::Entry::Entry() {}
81
Entry(uint64_t identifier,IThreadPool::ITask::Ptr task)82 TaskQueue::Entry::Entry(uint64_t identifier, IThreadPool::ITask::Ptr task) : task(move(task)), identifier(identifier) {}
83
operator ==(uint64_t rhsIdentifier) const84 bool TaskQueue::Entry::operator==(uint64_t rhsIdentifier) const
85 {
86 return identifier == rhsIdentifier;
87 }
88
operator ==(const TaskQueue::Entry & other) const89 bool TaskQueue::Entry::operator==(const TaskQueue::Entry& other) const
90 {
91 return identifier == other.identifier;
92 }
93 CORE_END_NAMESPACE()
94