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 #include <pthread.h>
16 #include <random>
17 #include "ffrt_inner.h"
18 #include "core/task_io.h"
19 #ifdef FFRT_CO_BACKTRACE_OH_ENABLE
20 #include <dlfcn.h>
21 #include "libunwind.h"
22 #endif
23 #include "util/slab.h"
24 #include "util/ffrt_facade.h"
25 #ifdef FFRT_IO_TASK_SCHEDULER
26 #include "util/spmc_queue.h"
27
28 #define ENABLE_LOCAL_QUEUE
29
30 namespace {
31 const int INSERT_GLOBAL_QUEUE_FREQ = 5;
32 }
33
34 namespace ffrt {
work_finish_callable(ffrt_executor_io_task * task)35 static void work_finish_callable(ffrt_executor_io_task* task)
36 {
37 task->status = ExecTaskStatus::ET_FINISH;
38 task->work.destroy(task->work.data);
39
40 #ifdef FFRT_BBOX_ENABLE
41 TaskDoneCounterInc();
42 #endif
43
44 delete task;
45 }
46
io_ffrt_executor_task_func(ffrt_executor_task_t * data,ffrt_qos_t qos)47 static void io_ffrt_executor_task_func(ffrt_executor_task_t* data, ffrt_qos_t qos)
48 {
49 ffrt_executor_io_task* task = static_cast<ffrt_executor_io_task*>(data);
50 task->status = ExecTaskStatus::ET_EXECUTING;
51 ffrt_coroutine_ptr_t coroutine = task->work.exec;
52 ffrt_coroutine_ret_t ret = coroutine(task->work.data);
53 if (ret == ffrt_coroutine_ready) {
54 FFRT_EXECUTOR_TASK_FINISH_MARKER(task);
55 work_finish_callable(task);
56 return;
57 }
58
59 FFRT_EXECUTOR_TASK_BLOCK_MARKER(task);
60 task->status = ffrt::ExecTaskStatus::ET_PENDING;
61 #ifdef FFRT_BBOX_ENABLE
62 TaskPendingCounterInc();
63 #endif
64 }
65
66 static pthread_once_t once = PTHREAD_ONCE_INIT;
67
ffrt_executor_io_task_init()68 static void ffrt_executor_io_task_init()
69 {
70 ffrt_executor_task_register_func(io_ffrt_executor_task_func, ffrt_io_task);
71 }
72 } /* namespace ffrt */
73
74 #ifdef __cplusplus
75 extern "C" {
76 #endif
77
78 API_ATTRIBUTE((visibility("default")))
ffrt_submit_coroutine(void * co,ffrt_coroutine_ptr_t exec,ffrt_function_t destroy,const ffrt_deps_t * in_deps,const ffrt_deps_t * out_deps,const ffrt_task_attr_t * attr)79 void ffrt_submit_coroutine(void* co, ffrt_coroutine_ptr_t exec, ffrt_function_t destroy,
80 const ffrt_deps_t* in_deps, const ffrt_deps_t* out_deps, const ffrt_task_attr_t* attr)
81 {
82 FFRT_COND_DO_ERR((exec==nullptr), return, "input invalid, exec==nullptr");
83 pthread_once(&ffrt::once, ffrt::ffrt_executor_io_task_init);
84
85 ffrt::task_attr_private *p = reinterpret_cast<ffrt::task_attr_private *>(const_cast<ffrt_task_attr_t *>(attr));
86 ffrt::QoS qos = (p == nullptr ? ffrt::QoS() : ffrt::QoS(p->qos_map));
87
88 ffrt::ffrt_executor_io_task* task = new ffrt::ffrt_executor_io_task(qos);
89 task->work.exec = exec;
90 task->work.destroy = destroy;
91 task->work.data = co;
92 task->status = ffrt::ExecTaskStatus::ET_READY;
93
94 ffrt_executor_task_submit((ffrt_executor_task_t*)task, attr);
95 }
96
97 API_ATTRIBUTE((visibility("default")))
ffrt_get_current_task()98 void* ffrt_get_current_task()
99 {
100 return reinterpret_cast<void*>(ffrt::ExecuteCtx::Cur()->exec_task);
101 }
102
103 // API used to schedule stackless coroutine task
104 API_ATTRIBUTE((visibility("default")))
ffrt_wake_coroutine(void * task)105 void ffrt_wake_coroutine(void* task)
106 {
107 if (task == nullptr) {
108 FFRT_LOGE("Task is nullptr");
109 return;
110 }
111
112 #ifdef FFRT_BBOX_ENABLE
113 TaskWakeCounterInc();
114 #endif
115
116 ffrt::ffrt_executor_io_task* wakedTask = static_cast<ffrt::ffrt_executor_io_task*>(task);
117 wakedTask->status = ffrt::ExecTaskStatus::ET_READY;
118
119 // in self-wakeup scenario, tasks are placed in local fifo to delay scheduling, implementing the veild funtion
120 bool selfAwake = (ffrt::ExecuteCtx::Cur()->exec_task == task);
121 if (!selfAwake && ffrt::ExecuteCtx::Cur()->PushTaskToPriorityStack(wakedTask)) {
122 return;
123 }
124
125 if (selfAwake || rand() % INSERT_GLOBAL_QUEUE_FREQ) {
126 if (ffrt::ExecuteCtx::Cur()->localFifo != nullptr &&
127 ffrt::ExecuteCtx::Cur()->localFifo->PushTail(task) == 0) {
128 ffrt::ExecuteUnit::Instance().NotifyLocalTaskAdded(wakedTask->qos);
129 return;
130 }
131 }
132
133 ffrt::LinkedList* node = (ffrt::LinkedList *)(&wakedTask->wq);
134 if (!ffrt::FFRTScheduler::Instance()->InsertNode(node, wakedTask->qos)) {
135 FFRT_LOGE("Submit IO task failed");
136 }
137 }
138
139 API_ATTRIBUTE((visibility("default")))
ffrt_task_attr_set_coroutine_type(ffrt_task_attr_t * attr,ffrt_coroutine_t coroutine_type)140 void ffrt_task_attr_set_coroutine_type(ffrt_task_attr_t* attr, ffrt_coroutine_t coroutine_type)
141 {
142 if (!attr) {
143 FFRT_LOGE("attr should be a valid address");
144 return;
145 }
146 return;
147 }
148
149 API_ATTRIBUTE((visibility("default")))
ffrt_task_attr_get_coroutine_type(const ffrt_task_attr_t * attr)150 ffrt_coroutine_t ffrt_task_attr_get_coroutine_type(const ffrt_task_attr_t* attr)
151 {
152 if (!attr) {
153 FFRT_LOGE("attr should be a valid address");
154 return ffrt_coroutine_with_stack;
155 }
156 return ffrt_coroutine_stackless;
157 }
158 #ifdef __cplusplus
159 }
160 #endif
161
162 #endif