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