applications.py 3.39 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
    def update_phase_context(self, msg):
        """Update the phase contextual information."""
64 65
        id = msg.cpu
        self.phase_contexts[id] = {k: getattr(msg, k) for k in ('startcompute',
66 67 68
                                   'endcompute', 'startbarrier', 'endbarrier')}
        self.phase_contexts[id]['set'] = True

69 70 71 72 73 74 75 76

class ApplicationManager(object):

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

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

77
    def register(self, msg, container):
78 79
        """Register a new downstream application."""

80 81 82 83
        uuid = msg.application_uuid
        container_uuid = msg.container_uuid
        progress = 0
        threads = False
84 85 86
        phase_contexts = dict()
        phase_context_keys = ['set', 'startcompute', 'endcompute',
                              'startbarrier', 'endbarrier']
87
        if container.power['policy']:
88 89 90 91 92 93 94 95
            ids = container.resources['cpus']
            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)
96 97 98 99

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