Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
gem5_logger.debug("gem5_shell output: {}".format(output))
# We get a second prompt. Hence, we need to eat one to make sure that we
# stay in sync. If we do not do this, we risk getting out of sync for
# slower simulations.
self.conn.expect([self.conn.UNIQUE_PROMPT, self.conn.PROMPT], timeout=self.default_timeout)
if check_exit_code:
exit_code_text = self._gem5_shell('echo $?', as_root=as_root,
timeout=timeout, check_exit_code=False,
sync=False)
try:
exit_code = int(exit_code_text.split()[0])
if exit_code:
message = 'Got exit code {}\nfrom: {}\nOUTPUT: {}'
raise TargetStableError(message.format(exit_code, command, output))
except (ValueError, IndexError):
gem5_logger.warning('Could not get exit code for "{}",\ngot: "{}"'.format(command, exit_code_text))
return output
with open(self.stderr_filename, 'r') as f:
for line in f:
# Look for two different strings, exact wording depends on
# version of gem5
m = re.search(r"Listening for system connection on port (?P\d+)", line)
if not m:
m = re.search(r"Listening for connections on port (?P\d+)", line)
if m:
port = int(m.group('port'))
if port >= 3456 and port < 5900:
self.gem5_port = port
break
# Check if the sockets are not disabled
m = re.search(r"Sockets disabled, not accepting terminal connections", line)
if m:
raise TargetStableError("The sockets have been disabled!"
"Pass --listener-mode=on to gem5")
else:
time.sleep(1)
raise TargetStableError(
"cannot split reply (target misconfiguration?):\n'{}'".format(full_output))
if check_exit_code:
try:
exit_code = int(exit_code_text)
if exit_code:
message = 'Got exit code {}\nfrom: {}\nOUTPUT: {}'
raise TargetStableError(message.format(exit_code, command, output))
except (ValueError, IndexError):
logger.warning(
'Could not get exit code for "{}",\ngot: "{}"'\
.format(command, exit_code_text))
return output
except EOF:
raise TargetNotRespondingError('Connection lost.')
except TargetStableError as e:
if will_succeed:
raise TargetTransientError(e)
else:
raise
# characters such as wildcard which would defeat the entire purpose of
# that function.
for c in [' ', "'", '"']:
pattern = pattern.replace(c, '\\' + c)
cmd = "exec printf '%s\n' {}".format(pattern)
# Make sure to use the same shell everywhere for the path globbing,
# ensuring consistent results no matter what is the default platform
# shell
cmd = '{} sh -c {} 2>/dev/null'.format(quote(self.busybox), quote(cmd))
# On some shells, match failure will make the command "return" a
# non-zero code, even though the command was not actually called
result = self.execute(cmd, strip_colors=False, check_exit_code=False, **kwargs)
paths = result.splitlines()
if not paths:
raise TargetStableError('No file matching: {}'.format(pattern))
return paths
def __init__(self, target):
if not hasattr(target, 'hwmon'):
raise TargetStableError('Target does not support HWMON')
super(HwmonInstrument, self).__init__(target)
self.logger.debug('Discovering available HWMON sensors...')
for ts in self.target.hwmon.sensors:
try:
ts.get_file('input')
measure = self.measure_map.get(ts.kind)[0]
if measure:
self.logger.debug('\tAdding sensor {}'.format(ts.name))
self.add_channel(_guess_site(ts), measure, sensor=ts)
else:
self.logger.debug('\tSkipping sensor {} (unknown kind "{}")'.format(ts.name, ts.kind))
except ValueError:
message = 'Skipping sensor {} because it does not have an input file'
self.logger.debug(message.format(ts.name))
continue
def write_value(self, path, value, verify=True):
value = str(value)
self.execute('echo {} > {}'.format(quote(value), quote(path)), check_exit_code=False, as_root=True)
if verify:
output = self.read_value(path)
if not output == value:
message = 'Could not set the value of {} to "{}" (read "{}")'.format(path, value, output)
raise TargetStableError(message)
def __init__(self, target, power_sites):
'''
Parameter power_sites is a list of gem5 identifiers for power values.
One example of such a field:
system.cluster0.cores0.power_model.static_power
'''
if not isinstance(target.platform, Gem5SimulationPlatform):
raise TargetStableError('Gem5PowerInstrument requires a gem5 platform')
if not target.has('gem5stats'):
raise TargetStableError('Gem5StatsModule is not loaded')
super(Gem5PowerInstrument, self).__init__(target)
# power_sites is assumed to be a list later
if isinstance(power_sites, list):
self.power_sites = power_sites
else:
self.power_sites = [power_sites]
self.add_channel('timestamp', 'time')
for field in self.power_sites:
self.add_channel(field, 'power')
self.target.gem5stats.book_roi(self.roi_label)
self.sample_period_ns = 10000000
# Sample rate must remain unset as gem5 does not provide samples
# at regular intervals therefore the reported timestamp should be used.
self.sample_rate_hz = None
self.target.gem5stats.start_periodic_dump(0, self.sample_period_ns)
def set_governor(self, governor):
if governor not in self.governors:
raise TargetStableError('Governor {} not supported for gpu'.format(governor))
self.target.write_value("/sys/kernel/gpu/gpu_governor", governor)
def _execute_command(self, command, as_root, log, timeout, executor):
# As we're already root, there is no need to use sudo.
log_debug = logger.debug if log else lambda msg: None
use_sudo = as_root and not self.connected_as_root
if use_sudo:
if self._sudo_needs_password and not self.password:
raise TargetStableError('Attempt to use sudo but no password was specified')
command = self.sudo_cmd.format(quote(command))
log_debug(command)
streams = executor(command, timeout=timeout)
if self._sudo_needs_password:
stdin = streams[0]
stdin.write(self.password + '\n')
stdin.flush()
else:
log_debug(command)
streams = executor(command, timeout=timeout)
return streams
def _prepare_xfer(self, action, sources, dest):
"""
Check the sanity of sources and destination and prepare the ground for
transfering multiple sources.
"""
if action == 'push':
src_excep = HostError
dst_excep = TargetStableError
dst_path_exists = self.file_exists
dst_is_dir = self.directory_exists
dst_mkdir = self.makedirs
for source in sources:
if not os.path.exists(source):
raise HostError('No such file "{}"'.format(source))
else:
src_excep = TargetStableError
dst_excep = HostError
dst_path_exists = os.path.exists
dst_is_dir = os.path.isdir
dst_mkdir = functools.partial(os.makedirs, exist_ok=True)
if not sources:
raise src_excep('No file matching: {}'.format(source))