applications.py 3.45 KB
Newer Older
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20
from __future__ import print_function

import logging

logger = logging.getLogger('nrm')


class Application(object):

    """Information about a downstream API user."""

    thread_fsm_table = {'stable': {'i': 's_ask_i', 'd': 's_ask_d'},
                        's_ask_i': {'done': 'stable', 'noop': 'max'},
                        's_ask_d': {'done': 'stable', 'noop': 'min'},
                        'max': {'d': 'max_ask_d'},
                        'min': {'i': 'min_ask_i'},
                        'max_ask_d': {'done': 'stable', 'noop': 'noop'},
                        'min_ask_i': {'done': 'stable', 'noop': 'noop'},
                        'noop': {}}

21
    def __init__(self, uuid, container, progress, threads, phase_contexts):
22 23 24 25 26
        self.uuid = uuid
        self.container_uuid = container
        self.progress = progress
        self.threads = threads
        self.thread_state = 'stable'
27
        self.phase_contexts = phase_contexts
28 29 30 31 32 33 34 35 36 37

    def do_thread_transition(self, event):
        """Update the thread fsm state."""
        transitions = self.thread_fsm_table[self.thread_state]
        if event in transitions:
            self.thread_state = transitions[event]

    def get_allowed_thread_requests(self):
        return self.thread_fsm_table[self.thread_state].keys()

38 39 40 41 42 43 44 45 46 47
    def get_thread_request_impact(self, command):
        # TODO: not a real model
        if command not in self.thread_fsm_table[self.thread_state]:
            return 0.0
        speed = float(self.progress)/float(self.threads['cur'])
        if command == 'i':
            return speed
        else:
            return -speed

48 49 50 51 52 53 54 55 56 57 58 59 60 61
    def update_threads(self, msg):
        """Update the thread tracking."""
        newth = msg['payload']
        curth = self.threads['cur']
        if newth == curth:
            self.do_thread_transition('noop')
        else:
            self.do_thread_transition('done')
        self.threads['cur'] = newth

    def update_progress(self, msg):
        """Update the progress tracking."""
        assert self.progress

62 63 64
    def update_performance(self, msg):
        """Update the progress tracking."""

65 66
    def update_phase_context(self, msg):
        """Update the phase contextual information."""
67 68 69 70
        id = int(msg.cpu)
        self.phase_contexts[id] = {k: getattr(msg, k) for k in
                                   ('aggregation', 'computetime',
                                    'totaltime')}
71 72
        self.phase_contexts[id]['set'] = True

73 74 75 76 77 78 79 80

class ApplicationManager(object):

    """Manages the tracking of applications: users of the downstream API."""

    def __init__(self):
        self.applications = dict()

81
    def register(self, msg, container):
82 83
        """Register a new downstream application."""

84 85 86 87
        uuid = msg.application_uuid
        container_uuid = msg.container_uuid
        progress = 0
        threads = False
88
        phase_contexts = dict()
89
        phase_context_keys = ['set', 'aggregation', 'computetime', 'totaltime']
90
        if container.power['policy']:
91
            ids = container.resources.cpus
92 93 94 95 96 97 98
            for id in ids:
                phase_contexts[id] = dict.fromkeys(phase_context_keys)
                phase_contexts[id]['set'] = False
        else:
            phase_contexts = None
        self.applications[uuid] = Application(uuid, container_uuid, progress,
                                              threads, phase_contexts)
99 100 101 102

    def delete(self, uuid):
        """Delete an application from the register."""
        del self.applications[uuid]