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
|
import logging
import os
from wa.framework import signal
from wa.framework.plugin import Parameter
from wa.framework.target.descriptor import (get_target_descriptions,
instantiate_target,
instantiate_assistant)
from wa.framework.target.info import TargetInfo
from wa.framework.target.runtime_parameter_manager import RuntimeParameterManager
from devlib import Gem5SimulationPlatform
from devlib.utils.misc import memoized
from devlib.exception import TargetError
class TargetManager(object):
"""
Instantiate the required target and perform configuration and validation of the device.
"""
parameters = [
Parameter('disconnect', kind=bool, default=False,
description="""
Specifies whether the target should be disconnected from
at the end of the run.
"""),
]
def __init__(self, name, parameters, outdir):
self.outdir = outdir
self.logger = logging.getLogger('tm')
self.target_name = name
self.target = None
self.assistant = None
self.platform_name = None
self.parameters = parameters
self.disconnect = parameters.get('disconnect')
self._init_target()
# If target supports hotplugging, online all cpus before perform discovery
# and restore original configuration after completed.
if self.target.has('hotplug'):
online_cpus = self.target.list_online_cpus()
try:
self.target.hotplug.online_all()
except TargetError:
msg = 'Failed to online all CPUS - some information may not be '\
'able to be retrieved.'
self.logger.debug(msg)
self.rpm = RuntimeParameterManager(self.target)
all_cpus = set(range(self.target.number_of_cpus))
self.target.hotplug.offline(*all_cpus.difference(online_cpus))
else:
self.rpm = RuntimeParameterManager(self.target)
def finalize(self):
if self.disconnect or isinstance(self.target.platform, Gem5SimulationPlatform):
self.logger.info('Disconnecting from the device')
with signal.wrap('TARGET_DISCONNECT'):
self.target.disconnect()
def start(self):
self.assistant.start()
def stop(self):
self.assistant.stop()
def extract_results(self, context):
self.assistant.extract_results(context)
@memoized
def get_target_info(self):
return TargetInfo(self.target)
def merge_runtime_parameters(self, parameters):
return self.rpm.merge_runtime_parameters(parameters)
def validate_runtime_parameters(self, parameters):
self.rpm.validate_runtime_parameters(parameters)
def commit_runtime_parameters(self, parameters):
self.rpm.commit_runtime_parameters(parameters)
def _init_target(self):
target_map = {td.name: td for td in get_target_descriptions()}
if self.target_name not in target_map:
raise ValueError('Unknown Target: {}'.format(self.target_name))
tdesc = target_map[self.target_name]
extra_plat_params={}
if tdesc.platform is Gem5SimulationPlatform:
extra_plat_params['host_output_dir'] = self.outdir
self.logger.debug('Creating {} target'.format(self.target_name))
self.target = instantiate_target(tdesc, self.parameters, connect=False,
extra_platform_params=extra_plat_params)
with signal.wrap('TARGET_CONNECT'):
self.target.connect()
self.logger.info('Setting up target')
self.target.setup()
self.assistant = instantiate_assistant(tdesc, self.parameters, self.target)
|
