Skip to content

property_setter

AdditionalSetter #

Bases: PropertySetterBase

Additional setter class.

Set additional parameters completely at random.

Source code in src/property_setter/additional_setter.py 
 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
class AdditionalSetter(PropertySetterBase):
    """Additional setter class.

    Set additional parameters completely at random.

    """

    def __init__(self, config: Config) -> None:
        """Constructor.

        Parameters
        ----------
        config : Config
            Config.

        """
        super().__init__(config)

    def _validate_config(self, config: Config) -> None:
        pass

    def set(self, dag: nx.DiGraph) -> None:
        """Completely random set additional properties.

        Parameters
        ----------
        dag : nx.DiGraph
            DAG

        """
        if node_properties := self._config.node_properties:
            for param_name, option in node_properties.items():
                for node_i in dag.nodes:
                    dag.nodes[node_i][param_name] = Util.random_choice(option)

        if edge_properties := self._config.edge_properties:
            for param_name, option in edge_properties.items():
                for src_i, tgt_i in dag.edges:
                    dag.edges[src_i, tgt_i][param_name] = Util.random_choice(option)

__init__(config) #

Constructor.

Parameters:

Name Type Description Default
config Config

Config.

 required
Source code in src/property_setter/additional_setter.py 
15
16
17
18
19
20
21
22
23
24
def __init__(self, config: Config) -> None:
    """Constructor.

    Parameters
    ----------
    config : Config
        Config.

    """
    super().__init__(config)

set(dag) #

Completely random set additional properties.

Parameters:

Name Type Description Default
dag nx.DiGraph

DAG

 required
Source code in src/property_setter/additional_setter.py 
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
def set(self, dag: nx.DiGraph) -> None:
    """Completely random set additional properties.

    Parameters
    ----------
    dag : nx.DiGraph
        DAG

    """
    if node_properties := self._config.node_properties:
        for param_name, option in node_properties.items():
            for node_i in dag.nodes:
                dag.nodes[node_i][param_name] = Util.random_choice(option)

    if edge_properties := self._config.edge_properties:
        for param_name, option in edge_properties.items():
            for src_i, tgt_i in dag.edges:
                dag.edges[src_i, tgt_i][param_name] = Util.random_choice(option)

CCRSetter #

Bases: PropertySetterBase

CCR setter class.

Source code in src/property_setter/ccr_setter.py 
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
class CCRSetter(PropertySetterBase):
    """CCR setter class."""

    def __init__(self, config: Config) -> None:
        """Constructor.

        Parameters
        ----------
        config : Config
            Config.

        """
        super().__init__(config)

    def _validate_config(self, config: Config) -> None:
        execution_time = config.execution_time
        communication_time = config.communication_time
        if execution_time and communication_time:
            logger.warning(
                "Both 'Execution time' and 'Communication time' are specified, "
                "but 'Communication time' is determined based on 'Execution time' and 'CCR'. "
                "So, the range of 'Communication time' entered is ignored."
            )

    def set(self, dag: nx.DiGraph) -> None:
        """Set CCR.

        Parameters
        ----------
        dag : nx.DiGraph
            DAG.

        Notes
        -----
        If both 'Execution time' and 'Communication time' are specified,
        'Execution time' is given priority and
        'Communication time' is calculated based on 'Execution time' and 'CCR'
        (i.e., the range of 'Communication time' specified is ignored).

        """
        ccr = Util.random_choice(self._config.ccr)
        if self._config.execution_time:
            self._set_by_exec(dag, ccr)
        else:
            self._set_by_comm(dag, ccr)

    def _set_by_exec(self, dag: nx.DiGraph, ccr: float) -> None:
        # Set execution time
        sum_exec = 0
        for node_i in dag.nodes():
            exec = Util.random_choice(self._config.execution_time)
            dag.nodes[node_i]["execution_time"] = exec
            sum_exec += exec

        # Calculate sum_comm
        sum_comm = int(ccr * sum_exec)

        # Set communication time
        comm_grouping = self._grouping(sum_comm, dag.number_of_edges())
        if not comm_grouping:
            self._output_round_up_warning("Communication time", "CCR")
            comm_grouping = [1 for _ in range(dag.number_of_edges())]
        for edge, comm in zip(dag.edges(), comm_grouping):
            dag.edges[edge[0], edge[1]]["communication_time"] = comm

    def _set_by_comm(self, dag: nx.DiGraph, ccr: float) -> None:
        # Set communication time
        sum_comm = 0
        for src_i, tgt_i in dag.edges():
            comm = Util.random_choice(self._config.communication_time)
            dag.edges[src_i, tgt_i]["communication_time"] = comm
            sum_comm += comm

        # Calculate sum_exec
        sum_exec = int(sum_comm / ccr)

        # Set communication time
        exec_grouping = self._grouping(sum_exec, dag.number_of_nodes())
        if not exec_grouping:
            self._output_round_up_warning("Execution time", "CCR")
            exec_grouping = [1 for _ in range(dag.number_of_nodes())]
        for node_i, exec in zip(dag.nodes(), exec_grouping):
            dag.nodes[node_i]["execution_time"] = exec

__init__(config) #

Constructor.

Parameters:

Name Type Description Default
config Config

Config.

 required
Source code in src/property_setter/ccr_setter.py 
15
16
17
18
19
20
21
22
23
24
def __init__(self, config: Config) -> None:
    """Constructor.

    Parameters
    ----------
    config : Config
        Config.

    """
    super().__init__(config)

set(dag) #

Set CCR.

Parameters:

Name Type Description Default
dag nx.DiGraph

DAG.

 required
Notes#

If both 'Execution time' and 'Communication time' are specified, 'Execution time' is given priority and 'Communication time' is calculated based on 'Execution time' and 'CCR' (i.e., the range of 'Communication time' specified is ignored).

Source code in src/property_setter/ccr_setter.py 
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
def set(self, dag: nx.DiGraph) -> None:
    """Set CCR.

    Parameters
    ----------
    dag : nx.DiGraph
        DAG.

    Notes
    -----
    If both 'Execution time' and 'Communication time' are specified,
    'Execution time' is given priority and
    'Communication time' is calculated based on 'Execution time' and 'CCR'
    (i.e., the range of 'Communication time' specified is ignored).

    """
    ccr = Util.random_choice(self._config.ccr)
    if self._config.execution_time:
        self._set_by_exec(dag, ccr)
    else:
        self._set_by_comm(dag, ccr)

DeadlineSetter #

Bases: PropertySetterBase

Deadline setter class.

Source code in src/property_setter/deadline_setter.py 
 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
class DeadlineSetter(PropertySetterBase):
    """Deadline setter class."""

    def __init__(self, config: Config) -> None:
        """Constructor.

        Parameters
        ----------
        config : Config
            Config.

        """
        super().__init__(config)

    def _validate_config(self, config: Config) -> None:
        pass

    def set(self, dag: nx.DiGraph) -> None:
        """Set end-to-end deadline based on critical path length.

        Parameters
        ----------
        dag : nx.DiGraph
            DAG.

        """
        for exit_i in Util.get_exit_nodes(dag):
            max_cp_len = 0
            for entry_i in Util.get_entry_nodes(dag):
                cp_len = self._get_cp_len(dag, entry_i, exit_i)
                if cp_len > max_cp_len:
                    max_cp_len = cp_len

            dag.nodes[exit_i]["end_to_end_deadline"] = int(
                max_cp_len * Util.random_choice(self._config.ratio_of_deadline_to_critical_path)
            )

    @staticmethod
    def _get_cp_len(dag: nx.DiGraph, source: int, exit: int) -> int:
        """Get critical path length.

        Parameters
        ----------
        dag : nx.DiGraph
            DAG.
        source : int
            Index of path source.
        exit : int
            Index of path exit.

        Returns
        -------
        int
            Critical path length.

        Notes
        -----
        If the edge has 'Communication time',
        'Communication time' is also included in critical path length.

        """
        cp_len = 0
        for path in nx.all_simple_paths(dag, source=source, target=exit):
            path_len = 0
            for i in range(len(path)):
                path_len += dag.nodes[path[i]]["execution_time"]
                if i != len(path) - 1 and dag.edges[path[i], path[i + 1]].get(
                    "communication_time"
                ):
                    path_len += dag.edges[path[i], path[i + 1]]["communication_time"]

            if path_len > cp_len:
                cp_len = path_len

        return cp_len

__init__(config) #

Constructor.

Parameters:

Name Type Description Default
config Config

Config.

 required
Source code in src/property_setter/deadline_setter.py 
11
12
13
14
15
16
17
18
19
20
def __init__(self, config: Config) -> None:
    """Constructor.

    Parameters
    ----------
    config : Config
        Config.

    """
    super().__init__(config)

set(dag) #

Set end-to-end deadline based on critical path length.

Parameters:

Name Type Description Default
dag nx.DiGraph

DAG.

 required
Source code in src/property_setter/deadline_setter.py 
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
def set(self, dag: nx.DiGraph) -> None:
    """Set end-to-end deadline based on critical path length.

    Parameters
    ----------
    dag : nx.DiGraph
        DAG.

    """
    for exit_i in Util.get_exit_nodes(dag):
        max_cp_len = 0
        for entry_i in Util.get_entry_nodes(dag):
            cp_len = self._get_cp_len(dag, entry_i, exit_i)
            if cp_len > max_cp_len:
                max_cp_len = cp_len

        dag.nodes[exit_i]["end_to_end_deadline"] = int(
            max_cp_len * Util.random_choice(self._config.ratio_of_deadline_to_critical_path)
        )

PropertySetterFactory #

Property setter factory class.

Source code in src/property_setter/property_setter_factory.py 
 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
class PropertySetterFactory:
    """Property setter factory class."""

    @staticmethod
    def create_utilization_setter(config: Config) -> UtilizationSetter:
        """Create utilization setter.

        Parameters
        ----------
        config : Config
            Config.

        Returns
        -------
        UtilizationSetter
            Utilization setter.

        """
        return UtilizationSetter(config)

    @staticmethod
    def create_ccr_setter(config: Config) -> CCRSetter:
        """Create CCR setter.

        Parameters
        ----------
        config : Config
            Config.

        Returns
        -------
        CCRSetter
            CCR setter.

        """
        return CCRSetter(config)

    @staticmethod
    def create_deadline_setter(config: Config) -> DeadlineSetter:
        """Create deadline setter.

        Parameters
        ----------
        config : Config
            Config.

        Returns
        -------
        DeadlineSetter
            Deadline setter.

        """
        return DeadlineSetter(config)

    @staticmethod
    def create_random_setter(config: Config, parameter_name: str, target: str) -> RandomSetter:
        """Create random setter.

        Parameters
        ----------
        config : Config
            Config
        parameter_name : str
            Parameter name
        target : str
            "node" or "edge"

        Returns
        -------
        RandomSetter
            Random setter.

        """
        return RandomSetter(config, parameter_name, target)

    @staticmethod
    def create_additional_setter(config: Config) -> AdditionalSetter:
        """Create additional setter.

        Parameters
        ----------
        config : Config
            Config.

        Returns
        -------
        AdditionalSetter
            Additional setter.

        """
        return AdditionalSetter(config)

create_additional_setter(config) staticmethod #

Create additional setter.

Parameters:

Name Type Description Default
config Config

Config.

 required

Returns:

Type Description
AdditionalSetter

Additional setter.
Source code in src/property_setter/property_setter_factory.py 
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
@staticmethod
def create_additional_setter(config: Config) -> AdditionalSetter:
    """Create additional setter.

    Parameters
    ----------
    config : Config
        Config.

    Returns
    -------
    AdditionalSetter
        Additional setter.

    """
    return AdditionalSetter(config)

create_ccr_setter(config) staticmethod #

Create CCR setter.

Parameters:

Name Type Description Default
config Config

Config.

 required

Returns:

Type Description
CCRSetter

CCR setter.
Source code in src/property_setter/property_setter_factory.py 
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
@staticmethod
def create_ccr_setter(config: Config) -> CCRSetter:
    """Create CCR setter.

    Parameters
    ----------
    config : Config
        Config.

    Returns
    -------
    CCRSetter
        CCR setter.

    """
    return CCRSetter(config)

create_deadline_setter(config) staticmethod #

Create deadline setter.

Parameters:

Name Type Description Default
config Config

Config.

 required

Returns:

Type Description
DeadlineSetter

Deadline setter.
Source code in src/property_setter/property_setter_factory.py 
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
@staticmethod
def create_deadline_setter(config: Config) -> DeadlineSetter:
    """Create deadline setter.

    Parameters
    ----------
    config : Config
        Config.

    Returns
    -------
    DeadlineSetter
        Deadline setter.

    """
    return DeadlineSetter(config)

create_random_setter(config, parameter_name, target) staticmethod #

Create random setter.

Parameters:

Name Type Description Default
config Config

Config

 required
parameter_name str

Parameter name

 required
target str

"node" or "edge"

 required

Returns:

Type Description
RandomSetter

Random setter.
Source code in src/property_setter/property_setter_factory.py 
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
@staticmethod
def create_random_setter(config: Config, parameter_name: str, target: str) -> RandomSetter:
    """Create random setter.

    Parameters
    ----------
    config : Config
        Config
    parameter_name : str
        Parameter name
    target : str
        "node" or "edge"

    Returns
    -------
    RandomSetter
        Random setter.

    """
    return RandomSetter(config, parameter_name, target)

create_utilization_setter(config) staticmethod #

Create utilization setter.

Parameters:

Name Type Description Default
config Config

Config.

 required

Returns:

Type Description
UtilizationSetter

Utilization setter.
Source code in src/property_setter/property_setter_factory.py 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
@staticmethod
def create_utilization_setter(config: Config) -> UtilizationSetter:
    """Create utilization setter.

    Parameters
    ----------
    config : Config
        Config.

    Returns
    -------
    UtilizationSetter
        Utilization setter.

    """
    return UtilizationSetter(config)

RandomSetter #

Bases: PropertySetterBase

Random setter class.

Set the specified parameters completely at random.

Source code in src/property_setter/random_setter.py 
 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
class RandomSetter(PropertySetterBase):
    """Random setter class.

    Set the specified parameters completely at random.

    """

    def __init__(self, config: Config, parameter_name: str, target: str) -> None:
        """Constructor.

        Parameters
        ----------
        config : Config
            Config.
        parameter_name : str
            Parameter name.
        target : str
            Target to be set.
            "node" or "edge".

        """
        super().__init__(config)
        self._param_name = parameter_name
        self._target = target

    def _validate_config(self, config: Config) -> None:
        pass

    def set(self, dag: nx.DiGraph) -> None:
        """Completely random set.

        Parameters
        ----------
        dag : nx.DiGraph
            DAG

        """
        property_name = Util.convert_to_property(self._param_name)
        option = getattr(self._config, property_name)
        if self._target == "node":
            for node_i in dag.nodes():
                dag.nodes[node_i][property_name] = Util.random_choice(option)
        else:
            for src_i, tgt_i in dag.edges():
                dag.edges[src_i, tgt_i][property_name] = Util.random_choice(option)

__init__(config, parameter_name, target) #

Constructor.

Parameters:

Name Type Description Default
config Config

Config.

 required
parameter_name str

Parameter name.

 required
target str

Target to be set. "node" or "edge".

 required
Source code in src/property_setter/random_setter.py 
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
def __init__(self, config: Config, parameter_name: str, target: str) -> None:
    """Constructor.

    Parameters
    ----------
    config : Config
        Config.
    parameter_name : str
        Parameter name.
    target : str
        Target to be set.
        "node" or "edge".

    """
    super().__init__(config)
    self._param_name = parameter_name
    self._target = target

set(dag) #

Completely random set.

Parameters:

Name Type Description Default
dag nx.DiGraph

DAG

 required
Source code in src/property_setter/random_setter.py 
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
def set(self, dag: nx.DiGraph) -> None:
    """Completely random set.

    Parameters
    ----------
    dag : nx.DiGraph
        DAG

    """
    property_name = Util.convert_to_property(self._param_name)
    option = getattr(self._config, property_name)
    if self._target == "node":
        for node_i in dag.nodes():
            dag.nodes[node_i][property_name] = Util.random_choice(option)
    else:
        for src_i, tgt_i in dag.edges():
            dag.edges[src_i, tgt_i][property_name] = Util.random_choice(option)

UtilizationSetter #

Bases: PropertySetterBase

Utilization setter class.

Notes#

If both 'Period' and 'Execution time' are specified, 'Execution time' is determined based on utilization and period (i.e., the range of 'Execution time' specified is ignored). The minimum value of 'Execution time' is 1 and never 0.

Source code in src/property_setter/utilization_setter.py 
 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
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
class UtilizationSetter(PropertySetterBase):
    """Utilization setter class.

    Notes
    -----
    If both 'Period' and 'Execution time' are specified,
    'Execution time' is determined based on utilization and period
    (i.e., the range of 'Execution time' specified is ignored).
    The minimum value of 'Execution time' is 1 and never 0.

    """

    def __init__(self, config: Config) -> None:
        super().__init__(config)

    def _validate_config(self, config: Config) -> None:
        period = config.period
        execution_time = config.execution_time
        if period and execution_time:
            logger.warning(
                "Both 'Period' and 'Execution time' are specified, "
                "but 'Execution time' is determined based on utilization and period. "
                "So, the range of 'Execution time' entered is ignored."
            )

    def set(self, dag: nx.DiGraph) -> None:
        """Set period and execution time based on utilization.

        Parameters
        ----------
        dag : nx.DiGraph
            DAG.

        Notes
        -----
        If a chain-based DAG is entered 'Periodic type' is 'Chain',
        the chain utilization is used (see https://par.nsf.gov/servlets/purl/10276465).

        """
        total_utilization = self._config.total_utilization
        is_chain_case = isinstance(dag, ChainBasedDAG) and Util.ambiguous_equals(
            self._config.periodic_type, "chain"
        )
        if is_chain_case:
            if total_utilization:
                self._set_by_total_utilization_chain(dag)
            else:
                self._set_by_only_max_utilization_chain(dag)
        else:
            if total_utilization:
                self._set_by_total_utilization(dag)
            else:
                self._set_by_only_max_utilization(dag)

        # Set remain execution times
        for node_i in dag.nodes:
            if not dag.nodes[node_i].get("execution_time"):
                dag.nodes[node_i]["execution_time"] = Util.random_choice(
                    self._config.execution_time
                )

    def _set_by_total_utilization(self, dag: nx.DiGraph) -> None:
        """Set period and execution time based on total utilization.

        Parameters
        ----------
        dag : nx.DiGraph
            DAG.

        """
        timer_driven_nodes = self._get_timer_driven_nodes(dag)
        utilizations = self._UUniFast(
            Util.random_choice(self._config.total_utilization),
            len(timer_driven_nodes),
            self._config.maximum_utilization,
        )

        for timer_i, utilization in zip(timer_driven_nodes, utilizations):
            selected_period = self._choice_period(dag, timer_i)
            dag.nodes[timer_i]["period"] = selected_period
            exec = int(utilization * selected_period)
            if exec == 0:
                self._output_round_up_warning("Execution time", "Utilization")
                exec = 1
            dag.nodes[timer_i]["execution_time"] = exec

    def _set_by_total_utilization_chain(self, chain_based_dag: ChainBasedDAG) -> None:
        timer_driven_nodes = self._get_timer_driven_nodes(chain_based_dag)
        utilizations = self._UUniFast(
            Util.random_choice(self._config.total_utilization),
            len(timer_driven_nodes),
            self._config.maximum_utilization,
        )

        for chain in chain_based_dag.chains:
            selected_period = self._choice_period(chain_based_dag, chain.head)
            chain_based_dag.nodes[chain.head]["period"] = selected_period
            utilization = utilizations[timer_driven_nodes.index(chain.head)]
            sum_exec = int(utilization * selected_period)
            exec_grouping = self._grouping(sum_exec, chain.number_of_nodes())  # type: ignore
            if not exec_grouping:
                self._output_round_up_warning("Execution time", "Utilization")
                exec_grouping = [1 for _ in range(chain.number_of_nodes())]
            for node_i, exec in zip(chain.nodes, exec_grouping):
                chain_based_dag.nodes[node_i]["execution_time"] = exec

    def _set_by_only_max_utilization(self, dag: nx.DiGraph) -> None:
        """Set period and execution time randomly by only maximum utilization.

        Parameters
        ----------
        dag : nx.DiGraph
            DAG.

        Notes
        -----
        If 'Maximum utilization' is not specified,
        'Maximum utilization' is set to 1.0.

        """
        max_u = self._config.maximum_utilization or 1.0
        for node_i in self._get_timer_driven_nodes(dag):
            selected_period = self._choice_period(dag, node_i)
            dag.nodes[node_i]["period"] = selected_period
            min_u = 1 / selected_period  # Ensure 'Execution time' is at least 1.
            if min_u > max_u:
                self._output_round_up_warning("Execution time", "Utilization")
                exec = 1
            else:
                utilization = random.uniform(min_u, max_u)
                exec = int(utilization * selected_period)
            dag.nodes[node_i]["execution_time"] = exec

    def _set_by_only_max_utilization_chain(self, chain_based_dag: ChainBasedDAG) -> None:
        """Set period and execution time randomly by only maximum utilization.

        Parameters
        ----------
        chain_based_dag: ChainBasedDAG
            Chain-based DAG.

        Notes
        -----
        If 'Maximum utilization' is not specified,
        'Maximum utilization' is set to 1.0.

        """
        max_u = self._config.maximum_utilization or 1.0
        for chain in chain_based_dag.chains:
            selected_period = self._choice_period(chain_based_dag, chain.head)
            chain_based_dag.nodes[chain.head]["period"] = selected_period
            min_u = (
                chain.number_of_nodes() / selected_period
            )  # Ensure 'Execution time' is at least 1.
            if min_u > max_u:
                self._output_round_up_warning("Execution time", "Utilization")
                exec_grouping = [1 for _ in range(chain.number_of_nodes())]
            else:
                utilization = random.uniform(min_u, max_u)
                sum_exec = int(utilization * selected_period)
                exec_grouping = self._grouping(sum_exec, chain.number_of_nodes())  # type: ignore
                if not exec_grouping:
                    self._output_round_up_warning("Execution time", "Utilization")
                    exec_grouping = [1 for _ in range(chain.number_of_nodes())]
            for node_i, exec in zip(chain.nodes, exec_grouping):
                chain_based_dag.nodes[node_i]["execution_time"] = exec

    @staticmethod
    def _UUniFast(total_u: float, n: int, max_u: Optional[float] = None) -> List[float]:
        """Determine utilization based on UUniFast method.

        For detail, see https://idp.springer.com/authorize/casa?redirect_uri=https://link.springer.com/content/pdf/10.1007/s11241-005-0507-9.pdf&casa_token=ILaVXw6_1aUAAAAA:KEgQ8Iv70JXyNHj7hs11YvW2KRIPm89ab_1bILtRZFI5sBU1A7QGYaNDMshx4up16pA4W2gDohyAQmJqWyc.

        Parameters
        ----------
        total_u : float
            Total utilization.
        n : int
            Number of elements to distribute utilization.
        max_u : float
            Maximum utilization, by default None.

        Returns
        -------
        List[float]
            List of utilizations.

        Notes
        -----
        - If both 'Total utilization' and 'Maximum utilization' cannot be met,
          ignore 'Total utilization' and set each utilization to 'Maximum utilization'.
        - If $'total_u' / 'n' \simeq 'max_u'$,
          it takes an enormous amount of time to distribute them.
          Therefore, if the number of attempts exceeds the threshold,
          the utilization is distributed equally.

        """
        if max_u:
            if (total_u / n) >= max_u:
                logger.warning(
                    "Only either 'Total utilization' or 'Maximum utilization' can be satisfied."
                    "Therefore, 'Total utilization' is ignored "
                    "and each utilization is set to 'Maximum utilization'."
                    "To prevent this, it is recommended to reduce 'Total utilization', "
                    "increase 'Maximum utilization', or increase the number of nodes."
                )
                utilizations = [max_u for _ in range(n)]
            else:
                utilizations = UtilizationSetter._UUniFast_with_max_u(total_u, n, max_u)

        else:  # Original UUniFast method
            remain_u = total_u
            utilizations: List[float] = []  # type: ignore
            for i in range(n - 1):
                next_u = -sys.maxsize
                next_u = remain_u * (random.uniform(0, 1) ** (1 / (n - i)))
                utilizations.append(remain_u - next_u)
                remain_u = next_u
            utilizations.append(remain_u)

        return utilizations

    @staticmethod
    def _UUniFast_with_max_u(total_u: float, n: int, max_u: float) -> List[float]:
        """Determine utilization based on UUniFast method not to exceed 'max_u'.

        Parameters
        ----------
        total_u : float
            Total utilization.
        n : int
            Number of elements to distribute utilization.
        max_u : float
            Maximum utilization.

        Returns
        -------
        List[float]
            List of utilizations.

        Notes
        -----
        If $'total_u' / 'n' \simeq 'max_u'$,
        it takes an enormous amount of time to distribute them.
        Therefore, if the number of attempts exceeds the threshold,
        the utilization is distributed equally.

        """
        max_try = 100  # HACK
        for try_i in range(1, max_try + 1):
            remain_u = total_u
            utilizations: List[float] = []
            for i in range(n - 1):
                next_u = -sys.maxsize
                while remain_u - next_u >= max_u:
                    next_u = remain_u * (random.uniform(0, 1) ** (1 / (n - i)))
                utilizations.append(remain_u - next_u)
                remain_u = next_u

            if remain_u < max_u:
                utilizations.append(remain_u)
                break

            if try_i == max_try:
                # HACK: Distribute equally.
                utilizations = [total_u / n for _ in range(n)]

        return utilizations

    def _choice_period(self, dag: nx.DiGraph, node_i: int) -> int:
        if self._config.entry_node_period and node_i in Util.get_entry_nodes(dag):
            return Util.random_choice(self._config.entry_node_period)
        if self._config.exit_node_period and node_i in Util.get_exit_nodes(dag):
            return Util.random_choice(self._config.exit_node_period)
        return Util.random_choice(self._config.period)

    def _get_timer_driven_nodes(self, dag: nx.DiGraph) -> List[int]:
        """Get indices of timer-driven nodes according to 'Periodic type'.

        Parameters
        ----------
        dag : nx.DiGraph
            DAG.

        Returns
        -------
        List[int]
            List of indexes of timer-driven nodes.

        """
        periodic_type = self._config.periodic_type
        timer_driven_nodes: List[int]
        if Util.ambiguous_equals(periodic_type, "All"):
            timer_driven_nodes = list(dag.nodes())
        elif Util.ambiguous_equals(periodic_type, "IO"):
            timer_driven_nodes = list(set(Util.get_entry_nodes(dag) + Util.get_exit_nodes(dag)))
        elif Util.ambiguous_equals(periodic_type, "Entry"):
            timer_driven_nodes = Util.get_entry_nodes(dag)
        elif isinstance(dag, ChainBasedDAG) and Util.ambiguous_equals(periodic_type, "Chain"):
            timer_driven_nodes = dag.chain_heads

        return timer_driven_nodes

set(dag) #

Set period and execution time based on utilization.

Parameters:

Name Type Description Default
dag nx.DiGraph

DAG.

 required
Notes#

If a chain-based DAG is entered 'Periodic type' is 'Chain', the chain utilization is used (see https://par.nsf.gov/servlets/purl/10276465).

Source code in src/property_setter/utilization_setter.py 
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
def set(self, dag: nx.DiGraph) -> None:
    """Set period and execution time based on utilization.

    Parameters
    ----------
    dag : nx.DiGraph
        DAG.

    Notes
    -----
    If a chain-based DAG is entered 'Periodic type' is 'Chain',
    the chain utilization is used (see https://par.nsf.gov/servlets/purl/10276465).

    """
    total_utilization = self._config.total_utilization
    is_chain_case = isinstance(dag, ChainBasedDAG) and Util.ambiguous_equals(
        self._config.periodic_type, "chain"
    )
    if is_chain_case:
        if total_utilization:
            self._set_by_total_utilization_chain(dag)
        else:
            self._set_by_only_max_utilization_chain(dag)
    else:
        if total_utilization:
            self._set_by_total_utilization(dag)
        else:
            self._set_by_only_max_utilization(dag)

    # Set remain execution times
    for node_i in dag.nodes:
        if not dag.nodes[node_i].get("execution_time"):
            dag.nodes[node_i]["execution_time"] = Util.random_choice(
                self._config.execution_time
            )