2 * Copyright (c) 2019 Samsung Electronics Co., Ltd. All Rights Reserved
4 * Licensed under the Apache License, Version 2.0 (the "License");
5 * you may not use this file except in compliance with the License.
6 * You may obtain a copy of the License at
8 * http://www.apache.org/licenses/LICENSE-2.0
10 * Unless required by applicable law or agreed to in writing, software
11 * distributed under the License is distributed on an "AS IS" BASIS,
12 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 * See the License for the specific language governing permissions and
14 * limitations under the License.
17 #include "ParallelExecutor.h"
21 #include "util/logging.h"
22 #include "exec/IFunction.h"
29 class HookFunction : public IFunction
32 HookFunction(IFunction *fn, const std::function<void()> &setup,
33 const std::function<void()> &teardown)
34 : _fn{fn}, _setup{setup}, _teardown{teardown}
48 std::function<void()> _setup;
49 std::function<void()> _teardown;
52 void ParallelExecutor::notify(uint32_t finished_job_id)
54 std::unique_lock<std::mutex> lock{_mu_jobs};
56 DataflowExecutor::notify(finished_job_id);
59 _cv_jobs.notify_all();
62 ParallelExecutor::ParallelExecutor(std::unique_ptr<compiler::LoweredGraph> lowered_graph,
63 const compiler::TensorRegistries &tensor_regs,
64 compiler::CodeMap &&code_map,
65 const util::TracingCtx *tracing_ctx)
66 : DataflowExecutor{std::move(lowered_graph), tensor_regs, std::move(code_map), tracing_ctx}
68 VERBOSE(ParallelExecutor) << "Constructing Parallel Executor" << std::endl;
71 void ParallelExecutor::executeImpl()
73 bool dynamic_input_exists = hasDynamicInput();
76 // TODO Consider to have distinct backend set in LowerInfoMap
78 for (auto &itr : _lowered_graph->getLowerInfo()->op_seq)
80 backends.add(itr.second->backend());
82 _scheduler = std::make_unique<ParallelScheduler>(backends);
84 assert(noWaitingJobs());
87 _waiting_jobs.swap(_finished_jobs); // Move finished jobs to waiting jobs
89 for (uint32_t i = 0; i < _waiting_jobs.size(); ++i)
91 VERBOSE(ParallelExecutor) << i << ": " << _input_info[i] << std::endl;
92 if (_input_info[i] == 0)
94 emplaceToReadyJobs(i);
97 assert(!_ready_jobs.empty()); // Cannot begin if there is no initial jobs
99 VERBOSE(ParallelExecutor) << "INITIAL JOBS : " << _ready_jobs.size() << std::endl;
101 auto profiling_subg_index = _tracing_ctx->getSubgraphIndex(&_graph);
103 _subject.notifySubgraphBegin(profiling_subg_index);
107 std::unique_lock<std::mutex> lock{_mu_jobs};
109 if (_ready_jobs.empty())
111 _cv_jobs.wait(lock, [this] { return !_ready_jobs.empty() || noWaitingJobs(); });
112 // Check finish condition
113 if (_ready_jobs.empty() && noWaitingJobs())
119 auto job = std::move(_ready_jobs.begin()->second);
120 _ready_jobs.erase(_ready_jobs.begin());
124 VERBOSE(ParallelExecutor) << "Assigning fn #" << job->index() << std::endl;
126 auto job_index = job->index();
127 auto op_sequence_index = _job_to_op_seq[job_index];
128 auto op_seq = &_lowered_graph->op_seqs().at(op_sequence_index);
129 auto backend = _lowered_graph->getLowerInfo()->op_seq.at(op_sequence_index)->backend();
130 auto setup = [&, op_seq, backend]() {
131 _subject.notifyJobBegin(this, profiling_subg_index, op_seq, backend);
133 auto teardown = [&, job_index, op_seq, backend]() {
134 _subject.notifyJobEnd(this, profiling_subg_index, op_seq, backend);
138 job->fn_seq()->initRunning();
140 // dynamic tensor setting
141 bool handle_dynamic_tensor = op_seq->has_dynamic_tensor() || dynamic_input_exists;
142 job->fn_seq()->enableDynamicShapeInferer(handle_dynamic_tensor);
144 _scheduler->assign(std::make_unique<HookFunction>(job->fn_seq(), setup, teardown), backend);
145 _finished_jobs[job_index] = std::move(job);
148 assert(noWaitingJobs());
150 // Wait for all the jobs done
151 _scheduler->finish();
152 _subject.notifySubgraphEnd(profiling_subg_index);
154 // Reset input info for the next execution
155 _input_info = _initial_input_info;