1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
|
# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements. See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from ....modeling import models
from .. import executor, api
class GraphCompiler(object):
def __init__(self, ctx, default_executor):
self._ctx = ctx
self._default_executor = default_executor
self._stub_executor = executor.base.StubTaskExecutor
self._model_to_api_id = {}
def compile(self,
task_graph,
start_stub_type=models.Task.START_WORKFLOW,
end_stub_type=models.Task.END_WORKFLOW,
depends_on=()):
"""
Translates the user graph to the execution graph
:param task_graph: The user's graph
:param start_stub_type: internal use
:param end_stub_type: internal use
:param depends_on: internal use
"""
depends_on = list(depends_on)
# Insert start marker
start_task = self._create_stub_task(
start_stub_type, depends_on, self._start_graph_suffix(task_graph.id), task_graph.name,
)
for task in task_graph.topological_order(reverse=True):
dependencies = \
(self._get_tasks_from_dependencies(task_graph.get_dependencies(task))
or [start_task])
if isinstance(task, api.task.OperationTask):
self._create_operation_task(task, dependencies)
elif isinstance(task, api.task.WorkflowTask):
# Build the graph recursively while adding start and end markers
self.compile(
task, models.Task.START_SUBWROFKLOW, models.Task.END_SUBWORKFLOW, dependencies
)
elif isinstance(task, api.task.StubTask):
self._create_stub_task(models.Task.STUB, dependencies, task.id)
else:
raise RuntimeError('Undefined state')
# Insert end marker
self._create_stub_task(
end_stub_type,
self._get_non_dependent_tasks(self._ctx.execution) or [start_task],
self._end_graph_suffix(task_graph.id),
task_graph.name
)
def _create_stub_task(self, stub_type, dependencies, api_id, name=None):
model_task = models.Task(
name=name,
dependencies=dependencies,
execution=self._ctx.execution,
_executor=self._stub_executor,
_stub_type=stub_type)
self._ctx.model.task.put(model_task)
self._model_to_api_id[model_task.id] = api_id
return model_task
def _create_operation_task(self, api_task, dependencies):
model_task = models.Task.from_api_task(
api_task, self._default_executor, dependencies=dependencies)
self._ctx.model.task.put(model_task)
self._model_to_api_id[model_task.id] = api_task.id
return model_task
@staticmethod
def _start_graph_suffix(api_id):
return '{0}-Start'.format(api_id)
@staticmethod
def _end_graph_suffix(api_id):
return '{0}-End'.format(api_id)
@staticmethod
def _get_non_dependent_tasks(execution):
tasks_with_dependencies = set()
for task in execution.tasks:
tasks_with_dependencies.update(task.dependencies)
return list(set(execution.tasks) - set(tasks_with_dependencies))
def _get_tasks_from_dependencies(self, dependencies):
"""
Returns task list from dependencies.
"""
tasks = []
for dependency in dependencies:
if isinstance(dependency, (api.task.StubTask, api.task.OperationTask)):
dependency_name = dependency.id
else:
dependency_name = self._end_graph_suffix(dependency.id)
tasks.extend(task for task in self._ctx.execution.tasks
if self._model_to_api_id.get(task.id, None) == dependency_name)
return tasks
|
