1 /*
2 * Copyright (c) 2021-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 "task_executor.h"
17
18 #include <pthread.h>
19 #include <thread>
20
21 #include "constants.h"
22 #include "dh_context.h"
23 #include "distributed_hardware_log.h"
24 #include "event_bus.h"
25
26 namespace OHOS {
27 namespace DistributedHardware {
28 namespace {
29 const uint32_t MAX_TASK_QUEUE_LENGTH = 256;
30 }
31 IMPLEMENT_SINGLE_INSTANCE(TaskExecutor);
TaskExecutor()32 TaskExecutor::TaskExecutor() : taskThreadFlag_(true)
33 {
34 DHLOGI("Ctor TaskExecutor");
35 std::thread(&TaskExecutor::TriggerTask, this).detach();
36 }
37
~TaskExecutor()38 TaskExecutor::~TaskExecutor()
39 {
40 DHLOGI("Dtor TaskExecutor");
41 taskThreadFlag_ = false;
42 }
43
PushTask(const std::shared_ptr<Task> & task)44 void TaskExecutor::PushTask(const std::shared_ptr<Task>& task)
45 {
46 if (task == nullptr) {
47 DHLOGE("Task is null");
48 return;
49 }
50
51 {
52 DHLOGI("Push task: %s", task->GetId().c_str());
53 std::unique_lock<std::mutex> lock(taskQueueMtx_);
54 if (taskQueue_.size() > MAX_TASK_QUEUE_LENGTH) {
55 DHLOGE("Task queue is full");
56 return;
57 }
58 taskQueue_.push(task);
59 }
60
61 condVar_.notify_one();
62 }
63
PopTask()64 std::shared_ptr<Task> TaskExecutor::PopTask()
65 {
66 std::shared_ptr<Task> task = nullptr;
67
68 std::unique_lock<std::mutex> lock(taskQueueMtx_);
69 condVar_.wait(lock, [this] {
70 return !(this->taskQueue_.empty());
71 });
72
73 if (!taskQueue_.empty()) {
74 task = taskQueue_.front();
75 taskQueue_.pop();
76 DHLOGI("Pop task: %s", task->GetId().c_str());
77 }
78
79 return task;
80 }
81
TriggerTask()82 void TaskExecutor::TriggerTask()
83 {
84 int32_t ret = pthread_setname_np(pthread_self(), TRIGGER_TASK);
85 if (ret != DH_FWK_SUCCESS) {
86 DHLOGE("TriggerTask setname failed.");
87 }
88 while (taskThreadFlag_) {
89 std::shared_ptr<Task> task = PopTask();
90 if (task == nullptr) {
91 DHLOGE("Pop a null task, error");
92 continue;
93 }
94
95 auto taskFunc = [task]() {
96 task->DoTask();
97 };
98
99 DHLOGI("Post task to EventBus: %s", task->GetId().c_str());
100 DHContext::GetInstance().GetEventBus()->PostTask(taskFunc, task->GetId());
101 }
102 }
103 } // namespace DistributedHardware
104 } // namespace OHOS
105