# Copyright 2013-2015 ARM Limited
#
# Licensed 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.
#
# pylint: disable=no-member
import logging
from copy import copy
from datetime import datetime
import wa.framework.signal as signal
from wa.framework import instrument
from wa.framework.configuration.core import Status
from wa.framework.exception import HostError, WorkloadError
from wa.framework.job import Job
from wa.framework.output import init_job_output
from wa.framework.output_processor import ProcessorManager
from wa.framework.resource import ResourceResolver
from wa.framework.target.manager import TargetManager
from wa.utils import log
from wa.utils.misc import merge_config_values, format_duration
class ExecutionContext(object):
@property
def previous_job(self):
if not self.job_queue:
return None
return self.job_queue[0]
@property
def next_job(self):
if not self.completed_jobs:
return None
return self.completed_jobs[-1]
@property
def spec_changed(self):
if self.previous_job is None and self.current_job is not None: # Start of run
return True
if self.previous_job is not None and self.current_job is None: # End of run
return True
return self.current_job.spec.id != self.previous_job.spec.id
@property
def spec_will_change(self):
if self.current_job is None and self.next_job is not None: # Start of run
return True
if self.current_job is not None and self.next_job is None: # End of run
return True
return self.current_job.spec.id != self.next_job.spec.id
@property
def workload(self):
if self.current_job:
return self.current_job.workload
@property
def job_output(self):
if self.current_job:
return self.current_job.output
@property
def output(self):
if self.current_job:
return self.job_output
return self.run_output
@property
def output_directory(self):
return self.output.basepath
def __init__(self, cm, tm, output):
self.logger = logging.getLogger('context')
self.cm = cm
self.tm = tm
self.run_output = output
self.run_state = output.state
self.target_info = self.tm.get_target_info()
self.logger.debug('Loading resource discoverers')
self.resolver = ResourceResolver(cm.plugin_cache)
self.resolver.load()
self.job_queue = None
self.completed_jobs = None
self.current_job = None
self.successful_jobs = 0
self.failed_jobs = 0
def start_run(self):
self.output.info.start_time = datetime.utcnow()
self.output.write_info()
self.job_queue = copy(self.cm.jobs)
self.completed_jobs = []
self.run_state.status = Status.STARTED
self.output.status = Status.STARTED
self.output.write_state()
def end_run(self):
if self.successful_jobs:
if self.failed_jobs:
status = Status.PARTIAL
else:
status = Status.OK
else:
status = Status.FAILED
self.run_state.status = status
self.run_output.status = status
self.run_output.info.end_time = datetime.utcnow()
self.run_output.info.duration = self.run_output.info.end_time -\
self.run_output.info.start_time
self.run_output.write_info()
self.run_output.write_state()
self.run_output.write_result()
def finalize(self):
self.tm.finalize()
def start_job(self):
if not self.job_queue:
raise RuntimeError('No jobs to run')
self.current_job = self.job_queue.pop(0)
job_output = init_job_output(self.run_output, self.current_job)
self.current_job.set_output(job_output)
self.update_job_state(self.current_job)
self.tm.start()
return self.current_job
def end_job(self):
if not self.current_job:
raise RuntimeError('No jobs in progress')
self.tm.stop()
self.completed_jobs.append(self.current_job)
self.update_job_state(self.current_job)
self.output.write_result()
self.current_job = None
def set_status(self, status, force=False):
if not self.current_job:
raise RuntimeError('No jobs in progress')
self.current_job.set_status(status, force)
def extract_results(self):
self.tm.extract_results(self)
def move_failed(self, job):
self.run_output.move_failed(job.output)
def update_job_state(self, job):
self.run_state.update_job(job)
self.run_output.write_state()
def skip_job(self, job):
job.status = Status.SKIPPED
self.run_state.update_job(job)
self.completed_jobs.append(job)
def skip_remaining_jobs(self):
while self.job_queue:
job = self.job_queue.pop(0)
self.skip_job(job)
self.write_state()
def write_state(self):
self.run_output.write_state()
def get_metric(self, name):
try:
return self.output.get_metric(name)
except HostError:
if not self.current_job:
raise
return self.run_output.get_metric(name)
def add_metric(self, name, value, units=None, lower_is_better=False,
classifiers=None):
if self.current_job:
classifiers = merge_config_values(self.current_job.classifiers,
classifiers)
self.output.add_metric(name, value, units, lower_is_better, classifiers)
def get_artifact(self, name):
try:
return self.output.get_artifact(name)
except HostError:
if not self.current_job:
raise
return self.run_output.get_artifact(name)
def get_artifact_path(self, name):
try:
return self.output.get_artifact_path(name)
except HostError:
if not self.current_job:
raise
return self.run_output.get_artifact_path(name)
def add_artifact(self, name, path, kind, description=None, classifiers=None):
self.output.add_artifact(name, path, kind, description, classifiers)
def add_run_artifact(self, name, path, kind, description=None,
classifiers=None):
self.run_output.add_artifact(name, path, kind, description, classifiers)
def add_event(self, message):
self.output.add_event(message)
def initialize_jobs(self):
new_queue = []
failed_ids = []
for job in self.job_queue:
if job.id in failed_ids:
# Don't try to initialize a job if another job with the same ID
# (i.e. same job spec) has failed - we can assume it will fail
# too.
self.skip_job(job)
continue
try:
job.initialize(self)
except WorkloadError as e:
job.set_status(Status.FAILED)
self.add_event(e.message)
if not getattr(e, 'logged', None):
log.log_error(e, self.logger)
e.logged = True
failed_ids.append(job.id)
if self.cm.run_config.bail_on_init_failure:
raise
else:
new_queue.append(job)
self.job_queue = new_queue
class Executor(object):
"""
The ``Executor``'s job is to set up the execution context and pass to a
``Runner`` along with a loaded run specification. Once the ``Runner`` has
done its thing, the ``Executor`` performs some final reporting before
returning.
The initial context set up involves combining configuration from various
sources, loading of requided workloads, loading and installation of
instruments and output processors, etc. Static validation of the combined
configuration is also performed.
"""
# pylint: disable=R0915
def __init__(self):
self.logger = logging.getLogger('executor')
self.error_logged = False
self.warning_logged = False
self.target_manager = None
self.device = None
def execute(self, config_manager, output):
"""
Execute the run specified by an agenda. Optionally, selectors may be
used to only selecute a subset of the specified agenda.
Params::
:state: a ``ConfigManager`` containing processed configuration
:output: an initialized ``RunOutput`` that will be used to
store the results.
"""
signal.connect(self._error_signalled_callback, signal.ERROR_LOGGED)
signal.connect(self._warning_signalled_callback, signal.WARNING_LOGGED)
self.logger.info('Initializing run')
self.logger.debug('Finalizing run configuration.')
config = config_manager.finalize()
output.write_config(config)
self.logger.info('Connecting to target')
self.target_manager = TargetManager(config.run_config.device,
config.run_config.device_config,
output.basepath)
output.set_target_info(self.target_manager.get_target_info())
self.logger.info('Initializing execution context')
context = ExecutionContext(config_manager, self.target_manager, output)
self.logger.info('Generating jobs')
config_manager.generate_jobs(context)
output.write_job_specs(config_manager.job_specs)
output.write_state()
self.logger.info('Installing instruments')
for instrument_name in config_manager.get_instruments(self.target_manager.target):
instrument.install(instrument_name, context)
instrument.validate()
self.logger.info('Installing output processors')
pm = ProcessorManager()
for proc in config_manager.get_processors():
pm.install(proc, context)
pm.validate()
self.logger.info('Starting run')
runner = Runner(context, pm)
signal.send(signal.RUN_STARTED, self)
runner.run()
context.finalize()
self.execute_postamble(context, output)
signal.send(signal.RUN_COMPLETED, self)
def execute_postamble(self, context, output):
self.logger.info('Done.')
duration = format_duration(output.info.duration)
self.logger.info('Run duration: {}'.format(duration))
num_ran = context.run_state.num_completed_jobs
status_summary = 'Ran a total of {} iterations: '.format(num_ran)
counter = context.run_state.get_status_counts()
parts = []
for status in reversed(Status.levels):
if status in counter:
parts.append('{} {}'.format(counter[status], status))
self.logger.info(status_summary + ', '.join(parts))
self.logger.info('Results can be found in {}'.format(output.basepath))
if self.error_logged:
self.logger.warn('There were errors during execution.')
self.logger.warn('Please see {}'.format(output.logfile))
elif self.warning_logged:
self.logger.warn('There were warnings during execution.')
self.logger.warn('Please see {}'.format(output.logfile))
def _error_signalled_callback(self):
self.error_logged = True
signal.disconnect(self._error_signalled_callback, signal.ERROR_LOGGED)
def _warning_signalled_callback(self):
self.warning_logged = True
signal.disconnect(self._warning_signalled_callback, signal.WARNING_LOGGED)
class Runner(object):
"""
Triggers running jobs and processing results
Takes pre-initialized ExcecutionContext and ProcessorManager. Handles
actually running the jobs, and triggers the ProcessorManager to handle
processing job and run results.
"""
def __init__(self, context, pm):
self.logger = logging.getLogger('runner')
self.logger.context = context
self.context = context
self.pm = pm
self.output = self.context.output
self.config = self.context.cm
def run(self):
try:
self.initialize_run()
self.send(signal.RUN_INITIALIZED)
while self.context.job_queue:
try:
with signal.wrap('JOB_EXECUTION', self, self.context):
self.run_next_job(self.context)
except KeyboardInterrupt:
self.context.skip_remaining_jobs()
except Exception as e:
self.context.add_event(e.message)
if (not getattr(e, 'logged', None) and
not isinstance(e, KeyboardInterrupt)):
log.log_error(e, self.logger)
e.logged = True
raise e
finally:
self.finalize_run()
self.send(signal.RUN_FINALIZED)
def initialize_run(self):
self.logger.info('Initializing run')
self.context.start_run()
self.pm.initialize()
log.indent()
self.context.initialize_jobs()
log.dedent()
self.context.write_state()
def finalize_run(self):
self.logger.info('Finalizing run')
self.context.end_run()
self.pm.enable_all()
self.pm.process_run_output(self.context)
self.pm.export_run_output(self.context)
self.pm.finalize()
log.indent()
for job in self.context.completed_jobs:
job.finalize(self.context)
log.dedent()
def run_next_job(self, context):
job = context.start_job()
self.logger.info('Running job {}'.format(job.id))
try:
log.indent()
self.do_run_job(job, context)
job.set_status(Status.OK)
except KeyboardInterrupt:
job.set_status(Status.ABORTED)
raise
except Exception as e: # pylint: disable=broad-except
job.set_status(Status.FAILED)
context.add_event(e.message)
if not getattr(e, 'logged', None):
log.log_error(e, self.logger)
e.logged = True
finally:
self.logger.info('Completing job {}'.format(job.id))
self.send(signal.JOB_COMPLETED)
context.end_job()
log.dedent()
self.check_job(job)
def do_run_job(self, job, context):
rc = self.context.cm.run_config
if job.workload.phones_home and not rc.allow_phone_home:
self.logger.warning('Skipping job {} ({}) due to allow_phone_home=False'
.format(job.id, job.workload.name))
self.context.skip_job(job)
return
job.set_status(Status.RUNNING)
self.send(signal.JOB_STARTED)
self.logger.info('Configuring augmentations')
job.configure_augmentations(context, self.pm)
with signal.wrap('JOB_TARGET_CONFIG', self, context):
job.configure_target(context)
with signal.wrap('JOB_SETUP', self, context):
job.setup(context)
try:
try:
with signal.wrap('JOB_EXECUTION', self, context):
job.run(context)
except Exception as e:
job.set_status(Status.FAILED)
if not getattr(e, 'logged', None):
log.log_error(e, self.logger)
e.logged = True
raise e
finally:
try:
with signal.wrap('JOB_OUTPUT_PROCESSED', self, context):
job.process_output(context)
self.pm.process_job_output(context)
self.pm.export_job_output(context)
except Exception:
job.set_status(Status.PARTIAL)
raise
except KeyboardInterrupt:
job.set_status(Status.ABORTED)
self.logger.info('Got CTRL-C. Aborting.')
raise
finally:
# If setup was successfully completed, teardown must
# run even if the job failed
with signal.wrap('JOB_TEARDOWN', self):
job.teardown(context)
def check_job(self, job):
rc = self.context.cm.run_config
if job.status in rc.retry_on_status:
if job.retries < rc.max_retries:
msg = 'Job {} iteration {} completed with status {}. retrying...'
self.logger.error(msg.format(job.id, job.iteration, job.status))
self.retry_job(job)
self.context.move_failed(job)
self.context.write_state()
else:
msg = 'Job {} iteration {} completed with status {}. '\
'Max retries exceeded.'
self.logger.error(msg.format(job.id, job.iteration, job.status))
self.context.failed_jobs += 1
else: # status not in retry_on_status
self.logger.info('Job completed with status {}'.format(job.status))
self.context.successful_jobs += 1
def retry_job(self, job):
retry_job = Job(job.spec, job.iteration, self.context)
retry_job.workload = job.workload
retry_job.retries = job.retries + 1
retry_job.set_status(Status.PENDING)
self.context.job_queue.insert(0, retry_job)
def send(self, s):
signal.send(s, self, self.context)
def __str__(self):
return 'runner'