applications.py 3.41 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
65
66
67
68
    def update_phase_context(self, msg):
        """Update the phase contextual information."""
        id = int(msg['cpu'])
        self.phase_contexts[id] = {k: int(msg[k]) for k in ('startcompute',
                                   '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
80
        """Register a new downstream application."""

        uuid = msg['uuid']
81
        container_uuid = msg['container']
82
83
        progress = msg['progress']
        threads = msg['threads']
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]