#include <DataflowExecutor.h>

Collaboration diagram for onert::exec::DataflowExecutor:

[legend]

Public Member Functions
	DataflowExecutor (std::unique_ptr< compiler::LoweredGraph > lowered_graph, backend::BackendContexts &&backend_contexts, const compiler::TensorRegistries &tensor_regs, compiler::CodeMap &&code_map, const util::TracingCtx *tracing_ctx)
	Constructs a DataflowExecutor object.

void	executeImpl (const ExecutionObservee &subject) override

Public Member Functions inherited from onert::exec::ExecutorBase
	ExecutorBase (std::unique_ptr< compiler::LoweredGraph > &&lowered_graph, backend::BackendContexts &&backend_contexts, const compiler::TensorRegistries &tensor_regs, const util::TracingCtx *tracing_ctx)
	Construct a new ExecutorBase object.

virtual	~ExecutorBase ()=default

const ir::Graph &	graph () const final
	Returns graph object.

void	execute (const std::vector< backend::IPortableTensor * > &inputs, const std::vector< backend::IPortableTensor * > &outputs, const ExecutionOptions &options) override
	Execute with given input/output tensors.

uint32_t	inputSize () const override
	Get input size.

uint32_t	outputSize () const override
	Get output size.

const ir::OperandInfo &	inputInfo (uint32_t index) const override
	Get input info at index.

const ir::OperandInfo &	outputInfo (uint32_t index) const override
	Get output info at index.

ir::Layout	inputLayout (uint32_t index) const override
	Get input layout at index.

ir::Layout	outputLayout (uint32_t index) const override
	Get output layout at index.

void	setIndexedRanks (std::shared_ptr< ir::OperationIndexMap< int64_t > > ranks) final
	Set an ordering on operations.

void	addObserver (std::unique_ptr< IExecutionObserver > ref)

backend::BackendContexts &	getBackendContexts ()

const ExecutionOptions &	currentOptions () const override
	Return current execution configuration.

Public Member Functions inherited from onert::exec::IExecutor
	IExecutor ()=default
	Construct a new IExecutor object.

virtual	~IExecutor ()=default
	Destroy the IExecutor object.

Protected Member Functions
virtual void	notify (uint32_t finished_job_id)

bool	noWaitingJobs ()

int64_t	calculateRank (const std::vector< ir::OperationIndex > &operations)

void	emplaceToReadyJobs (const uint32_t &id)

Protected Member Functions inherited from onert::exec::ExecutorBase
bool	hasDynamicInput ()
	Returns `true` if any input tensor is dynamic; `false` if all are static tensors.

Protected Attributes
compiler::CodeMap	_code_map

std::vector< std::unique_ptr< Job > >	_finished_jobs
	A vector of finished jobs for current execution After a run it has all the jobs of this execution for the next run.

std::vector< std::unique_ptr< Job > >	_waiting_jobs
	A vector of waiting jobs for current execution All the jobs are moved from _finished_jobs to it when start a run.

std::vector< std::list< uint32_t > >	_output_info
	Jobs' output info Used for notifying after finishing a job.

std::vector< uint32_t >	_initial_input_info

std::vector< uint32_t >	_input_info

std::multimap< int64_t, std::unique_ptr< Job >, std::greater< int64_t > >	_ready_jobs
	A collection of jobs that are ready for execution Jobs in it are ready to be scheduled. Ordered by priority from `_indexed_ranks`

std::unordered_map< uint32_t, ir::OperationIndex >	_job_to_op
	Which job runs which op and function.

Protected Attributes inherited from onert::exec::ExecutorBase
ExecObservers	_observers

std::shared_ptr< ir::OperationIndexMap< int64_t > >	_indexed_ranks

std::unique_ptr< compiler::LoweredGraph >	_lowered_graph

backend::BackendContexts	_backend_contexts

const ir::Graph &	_graph

std::vector< backend::builtin::IOTensor * >	_input_tensors

std::vector< backend::builtin::IOTensor * >	_output_tensors

std::mutex	_mutex

const util::TracingCtx *	_tracing_ctx

ExecutionOptions	_current_options

Detailed Description

Definition at line 35 of file DataflowExecutor.h.

Constructor & Destructor Documentation

◆ DataflowExecutor()

onert::exec::DataflowExecutor::DataflowExecutor	(	std::unique_ptr< compiler::LoweredGraph >	lowered_graph,
		backend::BackendContexts &&	backend_contexts,
		const compiler::TensorRegistries &	tensor_regs,
		compiler::CodeMap &&	code_map,
		const util::TracingCtx *	tracing_ctx
	)

Constructs a DataflowExecutor object.

Parameters

lowered_graph	LoweredGraph object
tensor_builders	Tensor builders that are currently used
code_map	`ir::Operation` and its code map

Definition at line 77 of file DataflowExecutor.cc.

  : ExecutorBase{std::move(lowered_graph), std::move(backend_contexts), tensor_regs, tracing_ctx},
    _code_map{std::move(code_map)}
{
  VERBOSE(DataflowExecutor) << "Constructing Dataflow Executor" << std::endl;
 
  // Assign jobs convert OperationIndex to job index(uint32_t)
  uint32_t next_job_index = 0;
  std::unordered_map<ir::OperationIndex, uint32_t> op_to_job;
  const auto &operations = _lowered_graph->graph().operations();
  operations.iterate([&](const ir::OperationIndex &op_ind, const ir::IOperation &) {
    VERBOSE(DataflowExecutor) << "Create a job " << next_job_index << " with Operation " << op_ind
                              << std::endl;
    _finished_jobs.emplace_back(
      std::make_unique<Job>(next_job_index, _code_map.at(op_ind).fn_seq.get()));
    op_to_job[op_ind] = next_job_index++;
  });
 
  _waiting_jobs.resize(next_job_index);
  _output_info.resize(next_job_index);
  _initial_input_info.resize(next_job_index, 0);
 
  operations.iterate([&](const ir::OperationIndex &op_ind, const ir::IOperation &op) {
    auto job_index = op_to_job[op_ind];
    for (auto &&output : op.getOutputs())
    {
      // Update output and input info
      operations.iterate([&](const ir::OperationIndex &op_cur_ind, const ir::IOperation &op_cur) {
        if (op_cur.getInputs().contains(output))
        {
          auto dep_index = op_to_job[op_cur_ind];
          ++_initial_input_info[dep_index];
          _output_info[job_index].push_back(dep_index);
        }
      });
    }
  });
  for (const auto &[op_ind, job_ind] : op_to_job)
    _job_to_op.emplace(job_ind, op_ind);
 
  _input_info = _initial_input_info;
}

References _code_map, _finished_jobs, _initial_input_info, _input_info, _job_to_op, onert::exec::ExecutorBase::_lowered_graph, _output_info, _waiting_jobs, onert::ir::OperandIndexSequence::contains(), onert::ir::IOperation::getInputs(), onert::ir::IOperation::getOutputs(), and VERBOSE.

Member Function Documentation

◆ calculateRank()

int64_t onert::exec::DataflowExecutor::calculateRank ( const std::vector< ir::OperationIndex > & operations )

protected

Definition at line 26 of file DataflowExecutor.cc.

{
  int64_t rank = 0;
  if (!_indexed_ranks)
  {
    return rank;
  }
  for (const auto &operation_idx : operations)
  {
    auto it = _indexed_ranks->find(operation_idx);
    if (it == _indexed_ranks->end())
    {
      assert(_graph.operations().at(operation_idx).opcode() == ir::OpCode::Permute &&
             operations.size() == 1);
      // run Permute ASAP for next operations to be ready for other backends
      return std::numeric_limits<int64_t>::max();
    }
    else
    {
      rank += it->second;
    }
  }
  return rank;
}

References onert::exec::ExecutorBase::_graph, onert::exec::ExecutorBase::_indexed_ranks, onert::util::ObjectManager< Index, Object >::at(), and onert::ir::Graph::operations().

Referenced by emplaceToReadyJobs().

◆ emplaceToReadyJobs()

void onert::exec::DataflowExecutor::emplaceToReadyJobs ( const uint32_t & id )

protected

Definition at line 51 of file DataflowExecutor.cc.

{
  auto &job = _waiting_jobs[id];
  assert(job != nullptr);
  auto rank = calculateRank({_job_to_op[job->index()]});
  _ready_jobs.emplace(rank, std::move(job));
}

References _job_to_op, _ready_jobs, _waiting_jobs, and calculateRank().

Referenced by executeImpl(), onert::exec::ParallelExecutor::executeImpl(), and notify().

◆ executeImpl()

void onert::exec::DataflowExecutor::executeImpl ( const ExecutionObservee & subject )

overridevirtual

Implements onert::exec::ExecutorBase.

Reimplemented in onert::exec::ParallelExecutor.

Definition at line 124 of file DataflowExecutor.cc.

{
  assert(noWaitingJobs());
 
  bool dynamic_input_exists = hasDynamicInput();
 
  // Execution setup
  _waiting_jobs.swap(_finished_jobs); // Move finished jobs to waiting jobs
 
  for (uint32_t i = 0; i < _waiting_jobs.size(); ++i)
  {
    if (_input_info[i] == 0)
    {
      emplaceToReadyJobs(i);
    }
  }
  assert(!_ready_jobs.empty()); // Cannot begin if there is no initial jobs
 
  auto profiling_subg_index = _tracing_ctx->getSubgraphIndex(&_graph);
 
  subject.notifySubgraphBegin(profiling_subg_index);
 
  while (!_ready_jobs.empty())
  {
    auto job = std::move((_ready_jobs.begin())->second);
    _ready_jobs.erase(_ready_jobs.begin());
    auto job_index = job->index();
    VERBOSE(DataflowExecutor) << "Run job " << job_index << std::endl;
 
    auto op_ind = _job_to_op[job_index];
    const auto backend = _lowered_graph->lower_info().operation.at(op_ind);
 
    subject.notifyJobBegin(this, profiling_subg_index, op_ind, backend);
 
    job->fn_seq()->initRunning();
 
    // check if FunctionSequence needs to handle dynamic tensor
    bool handle_dynamic_tensor =
      _lowered_graph->getHasDynamicTensor(op_ind) || dynamic_input_exists;
    job->fn_seq()->enableDynamicShapeInferer(handle_dynamic_tensor);
 
    job->run();
 
    subject.notifyJobEnd(this, profiling_subg_index, op_ind, backend);
    notify(job_index);
    _finished_jobs[job_index] = std::move(job);
  }
  assert(noWaitingJobs());
 
  subject.notifySubgraphEnd(profiling_subg_index);
 
  // Reset input info for the next execution
  _input_info = _initial_input_info;
}

◆ notify()

void onert::exec::DataflowExecutor::notify ( uint32_t finished_job_id )

protectedvirtual

Reimplemented in onert::exec::ParallelExecutor.

Definition at line 59 of file DataflowExecutor.cc.

{
  for (auto &&id : _output_info[finished_job_id])
  {
    assert(_input_info[id] > 0);
    auto count = --_input_info[id];
    if (count == 0) // No dependent jobs left, ready for execution
    {
      emplaceToReadyJobs(id);
    }
  }
}

References _input_info, _output_info, and emplaceToReadyJobs().

Referenced by executeImpl(), and onert::exec::ParallelExecutor::notify().

◆ noWaitingJobs()

bool onert::exec::DataflowExecutor::noWaitingJobs ( )

protected

Definition at line 71 of file DataflowExecutor.cc.

{
  return std::all_of(_waiting_jobs.begin(), _waiting_jobs.end(),
                     [](const std::unique_ptr<Job> &job) { return job == nullptr; });
}

References _waiting_jobs.

Referenced by executeImpl(), and onert::exec::ParallelExecutor::executeImpl().

Field Documentation

◆ _code_map

compiler::CodeMap onert::exec::DataflowExecutor::_code_map

protected

Definition at line 62 of file DataflowExecutor.h.

Referenced by DataflowExecutor().

◆ _finished_jobs

std::vector<std::unique_ptr<Job> > onert::exec::DataflowExecutor::_finished_jobs

protected

A vector of finished jobs for current execution After a run it has all the jobs of this execution for the next run.

Definition at line 67 of file DataflowExecutor.h.

Referenced by DataflowExecutor(), executeImpl(), and onert::exec::ParallelExecutor::executeImpl().

◆ _initial_input_info

std::vector<uint32_t> onert::exec::DataflowExecutor::_initial_input_info

protected

Definition at line 78 of file DataflowExecutor.h.

Referenced by DataflowExecutor(), executeImpl(), and onert::exec::ParallelExecutor::executeImpl().

◆ _input_info

std::vector<uint32_t> onert::exec::DataflowExecutor::_input_info

protected

Definition at line 79 of file DataflowExecutor.h.

Referenced by DataflowExecutor(), executeImpl(), onert::exec::ParallelExecutor::executeImpl(), and notify().

◆ _job_to_op

std::unordered_map<uint32_t, ir::OperationIndex> onert::exec::DataflowExecutor::_job_to_op

protected

Which job runs which op and function.

Definition at line 88 of file DataflowExecutor.h.

Referenced by DataflowExecutor(), emplaceToReadyJobs(), executeImpl(), and onert::exec::ParallelExecutor::executeImpl().

◆ _output_info

std::vector<std::list<uint32_t> > onert::exec::DataflowExecutor::_output_info

protected

Jobs' output info Used for notifying after finishing a job.

Definition at line 77 of file DataflowExecutor.h.

Referenced by DataflowExecutor(), and notify().

◆ _ready_jobs

std::multimap<int64_t, std::unique_ptr<Job>, std::greater<int64_t> > onert::exec::DataflowExecutor::_ready_jobs

protected

A collection of jobs that are ready for execution Jobs in it are ready to be scheduled. Ordered by priority from _indexed_ranks

Definition at line 85 of file DataflowExecutor.h.

Referenced by emplaceToReadyJobs(), executeImpl(), and onert::exec::ParallelExecutor::executeImpl().

◆ _waiting_jobs

std::vector<std::unique_ptr<Job> > onert::exec::DataflowExecutor::_waiting_jobs

protected

A vector of waiting jobs for current execution All the jobs are moved from _finished_jobs to it when start a run.

Definition at line 72 of file DataflowExecutor.h.

Referenced by DataflowExecutor(), emplaceToReadyJobs(), executeImpl(), onert::exec::ParallelExecutor::executeImpl(), and noWaitingJobs().

The documentation for this class was generated from the following files:

runtime/onert/core/src/exec/DataflowExecutor.h
runtime/onert/core/src/exec/DataflowExecutor.cc

Public Member Functions

Protected Member Functions

Protected Attributes

Detailed Description

Constructor & Destructor Documentation

◆ DataflowExecutor()

Member Function Documentation

◆ calculateRank()

◆ emplaceToReadyJobs()

◆ executeImpl()

◆ notify()

◆ noWaitingJobs()

Field Documentation

◆ _code_map

◆ _finished_jobs

◆ _initial_input_info

◆ _input_info

◆ _job_to_op

◆ _output_info

◆ _ready_jobs

◆ _waiting_jobs