#
# Author: Alina Quereilhac <alina.quereilhac@inria.fr>
-import functools
-import logging
-import os
-import random
-import sys
-import time
-import threading
-
from nepi.util import guid
from nepi.util.parallel import ParallelRun
from nepi.util.timefuncs import tnow, tdiffsec, stabsformat, tsformat
from nepi.execution.trace import TraceAttr
# TODO: use multiprocessing instead of threading
-# TODO: When a failure occurs during deployment, scp and ssh processes are left running behind!!
# TODO: Allow to reconnect to a running experiment instance! (reconnect mode vs deploy mode)
+import functools
+import logging
+import os
+import random
+import sys
+import time
+import threading
+
class ECState(object):
""" State of the Experiment Controller
def __init__(self, exp_id = None):
super(ExperimentController, self).__init__()
- # root directory to store files
+ # Logging
+ self._logger = logging.getLogger("ExperimentController")
# Run identifier. It identifies a concrete instance (run) of an experiment.
# Since a same experiment (same configuration) can be run many times,
# Tasks
self._tasks = dict()
+ # RM groups
+ self._groups = dict()
+
+ # generator of globally unique id for groups
+ self._group_id_generator = guid.GuidGenerator()
+
# Event processing thread
self._cond = threading.Condition()
self._thread = threading.Thread(target = self._process)
# EC state
self._state = ECState.RUNNING
- # Logging
- self._logger = logging.getLogger("ExperimentController")
-
@property
def logger(self):
""" Return the logger of the Experiment Controller
def wait_finished(self, guids):
""" Blocking method that wait until all the RM from the 'guid' list
- reached the state FINISHED
+ reached the state FINISHED ( or STOPPED, FAILED or RELEASED )
:param guids: List of guids
:type guids: list
def wait_started(self, guids):
""" Blocking method that wait until all the RM from the 'guid' list
- reached the state STARTED
+ reached the state STARTED ( or STOPPED, FINISHED, FAILED, RELEASED)
:param guids: List of guids
:type guids: list
"""
- return self.wait(guids, states = [ResourceState.STARTED,
- ResourceState.STOPPED,
- ResourceState.FINISHED])
+ return self.wait(guids, state = ResourceState.STARTED)
+
+ def wait_released(self, guids):
+ """ Blocking method that wait until all the RM from the 'guid' list
+ reached the state RELEASED (or FAILED)
- def wait(self, guids, states = [ResourceState.FINISHED,
- ResourceState.STOPPED]):
+ :param guids: List of guids
+ :type guids: list
+ """
+ # TODO: solve state concurrency BUG and !!!!
+ # correct waited release state to state = ResourceState.FAILED)
+ return self.wait(guids, state = ResourceState.FINISHED)
+
+ def wait_deployed(self, guids):
+ """ Blocking method that wait until all the RM from the 'guid' list
+ reached the state READY (or any higher state)
+
+ :param guids: List of guids
+ :type guids: list
+ """
+ return self.wait(guids, state = ResourceState.READY)
+
+ def wait(self, guids, state = ResourceState.STOPPED):
""" Blocking method that waits until all the RM from the 'guid' list
reached state 'state' or until a failure occurs
if isinstance(guids, int):
guids = [guids]
- while not all([self.state(guid) in states for guid in guids]) and \
- not any([self.state(guid) in [
- ResourceState.FAILED] for guid in guids]) and \
- not self.finished:
- # debug logging
- waited = ""
- for guid in guids:
- waited += "guid %d - %s \n" % (guid, self.state(guid, hr = True))
- self.logger.debug(" WAITING FOR %s " % waited )
-
- # We keep the sleep big to decrease the number of RM state queries
- time.sleep(2)
-
+ # we randomly alter the order of the guids to avoid ordering
+ # dependencies (e.g. LinuxApplication RMs runing on the same
+ # linux host will be synchronized by the LinuxNode SSH lock)
+ random.shuffle(guids)
+
+ while True:
+ # If no more guids to wait for or an error occured, then exit
+ if len(guids) == 0 or self.finished:
+ break
+
+ # If a guid reached one of the target states, remove it from list
+ guid = guids[0]
+ rstate = self.state(guid)
+
+ if rstate >= state:
+ guids.remove(guid)
+ else:
+ # Debug...
+ self.logger.debug(" WAITING FOR guid %d - state is %s, required is >= %s " % (guid,
+ self.state(guid, hr = True), state))
+
+ # Take the opportunity to 'refresh' the states of the RMs.
+ # Query only the first up to N guids (not to overwhelm
+ # the local machine)
+ n = 100
+ lim = n if len(guids) > n else ( len(guids) -1 )
+ nguids = guids[0: lim]
+
+ # schedule state request for all guids (take advantage of
+ # scheduler multi threading).
+ for guid in nguids:
+ callback = functools.partial(self.state, guid)
+ self.schedule("0s", callback)
+
+ # If the guid is not in one of the target states, wait and
+ # continue quering. We keep the sleep big to decrease the
+ # number of RM state queries
+ time.sleep(4)
+
def get_task(self, tid):
""" Get a specific task
rm1.register_connection(guid2)
rm2.register_connection(guid1)
- def register_condition(self, group1, action, group2, state,
+ def register_condition(self, guids1, action, guids2, state,
time = None):
- """ Registers an action START or STOP for all RM on group1 to occur
- time 'time' after all elements in group2 reached state 'state'.
+ """ Registers an action START or STOP for all RM on guids1 to occur
+ time 'time' after all elements in guids2 reached state 'state'.
- :param group1: List of guids of RMs subjected to action
- :type group1: list
+ :param guids1: List of guids of RMs subjected to action
+ :type guids1: list
:param action: Action to register (either START or STOP)
:type action: ResourceAction
- :param group2: List of guids of RMs to we waited for
- :type group2: list
+ :param guids2: List of guids of RMs to we waited for
+ :type guids2: list
:param state: State to wait for on RMs (STARTED, STOPPED, etc)
:type state: ResourceState
- :param time: Time to wait after group2 has reached status
+ :param time: Time to wait after guids2 has reached status
:type time: string
"""
- if isinstance(group1, int):
- group1 = [group1]
- if isinstance(group2, int):
- group2 = [group2]
+ if isinstance(guids1, int):
+ guids1 = [guids1]
+ if isinstance(guids2, int):
+ guids2 = [guids2]
- for guid1 in group1:
+ for guid1 in guids1:
rm = self.get_resource(guid1)
- rm.register_condition(action, group2, state, time)
+ rm.register_condition(action, guids2, state, time)
- def register_trace(self, guid, name):
+ def enable_trace(self, guid, name):
""" Enable trace
:param name: Name of the trace
:type name: str
"""
rm = self.get_resource(guid)
- rm.register_trace(name)
+ rm.enable_trace(name)
+
+ def trace_enabled(self, guid, name):
+ """ Returns True if trace is enabled
+
+ :param name: Name of the trace
+ :type name: str
+ """
+ rm = self.get_resource(guid)
+ return rm.trace_enabled(name)
def trace(self, guid, name, attr = TraceAttr.ALL, block = 512, offset = 0):
""" Get information on collected trace
rm = self.get_resource(guid)
return rm.start()
- def set_with_conditions(self, name, value, group1, group2, state,
+ def set_with_conditions(self, name, value, guids1, guids2, state,
time = None):
""" Set value 'value' on attribute with name 'name' on all RMs of
- group1 when 'time' has elapsed since all elements in group2
+ guids1 when 'time' has elapsed since all elements in guids2
have reached state 'state'.
:param name: Name of attribute to set in RM
:param value: Value of attribute to set in RM
:type name: string
- :param group1: List of guids of RMs subjected to action
- :type group1: list
+ :param guids1: List of guids of RMs subjected to action
+ :type guids1: list
:param action: Action to register (either START or STOP)
:type action: ResourceAction
- :param group2: List of guids of RMs to we waited for
- :type group2: list
+ :param guids2: List of guids of RMs to we waited for
+ :type guids2: list
:param state: State to wait for on RMs (STARTED, STOPPED, etc)
:type state: ResourceState
- :param time: Time to wait after group2 has reached status
+ :param time: Time to wait after guids2 has reached status
:type time: string
"""
- if isinstance(group1, int):
- group1 = [group1]
- if isinstance(group2, int):
- group2 = [group2]
+ if isinstance(guids1, int):
+ guids1 = [guids1]
+ if isinstance(guids2, int):
+ guids2 = [guids2]
- for guid1 in group1:
+ for guid1 in guids1:
rm = self.get_resource(guid)
- rm.set_with_conditions(name, value, group2, state, time)
+ rm.set_with_conditions(name, value, guids2, state, time)
def stop_with_conditions(self, guid):
""" Stop a specific RM defined by its 'guid' only if all the conditions are true
"""
rm = self.get_resource(guid)
- return rm.start_with_condition()
+ return rm.start_with_conditions()
- def deploy(self, group = None, wait_all_ready = True):
- """ Deploy all resource manager in group
+ def deploy(self, guids = None, wait_all_ready = True, group = None):
+ """ Deploy all resource manager in guids list
- :param group: List of guids of RMs to deploy
- :type group: list
+ :param guids: List of guids of RMs to deploy
+ :type guids: list
:param wait_all_ready: Wait until all RMs are ready in
order to start the RMs
:type guid: int
+ :param group: Id of deployment group in which to deploy RMs
+ :type group: int
+
"""
self.logger.debug(" ------- DEPLOY START ------ ")
- if not group:
- # By default, if not deployment group is indicated,
- # all RMs that are undeployed will be deployed
- group = []
+ if not guids:
+ # If no guids list was indicated, all 'NEW' RMs will be deployed
+ guids = []
for guid in self.resources:
if self.state(guid) == ResourceState.NEW:
- group.append(guid)
+ guids.append(guid)
- if isinstance(group, int):
- group = [group]
+ if isinstance(guids, int):
+ guids = [guids]
+
+ # Create deployment group
+ new_group = False
+ if not group:
+ new_group = True
+ group = self._group_id_generator.next(guid)
- # Before starting deployment we disorder the group list with the
+ if group not in self._groups:
+ self._groups[group] = []
+
+ self._groups[group].extend(guids)
+
+ # Before starting deployment we disorder the guids list with the
# purpose of speeding up the whole deployment process.
- # It is likely that the user inserted in the 'group' list closely
+ # It is likely that the user inserted in the 'guids' list closely
# resources one after another (e.g. all applications
# connected to the same node can likely appear one after another).
# This can originate a slow down in the deployment since the N
# be taken up by the same family of resources waiting for the
# same conditions (e.g. LinuxApplications running on a same
# node share a single lock, so they will tend to be serialized).
- # If we disorder the group list, this problem can be mitigated.
- random.shuffle(group)
+ # If we disorder the guids list, this problem can be mitigated.
+ random.shuffle(guids)
def wait_all_and_start(group):
reschedule = False
- for guid in group:
+
+ # Get all guids in group
+ guids = self._groups[group]
+
+ for guid in guids:
if self.state(guid) < ResourceState.READY:
reschedule = True
break
self.schedule("1s", callback)
else:
# If all resources are read, we schedule the start
- for guid in group:
+ for guid in guids:
rm = self.get_resource(guid)
self.schedule("0s", rm.start_with_conditions)
- if wait_all_ready:
- # Schedule the function that will check all resources are
- # READY, and only then it will schedule the start.
- # This is aimed to reduce the number of tasks looping in the scheduler.
- # Intead of having N start tasks, we will have only one
+ if wait_all_ready and new_group:
+ # Schedule a function to check that all resources are
+ # READY, and only then schedule the start.
+ # This aimes at reducing the number of tasks looping in the
+ # scheduler.
+ # Intead of having N start tasks, we will have only one for
+ # the whole group.
callback = functools.partial(wait_all_and_start, group)
self.schedule("1s", callback)
- for guid in group:
+ for guid in guids:
rm = self.get_resource(guid)
+ rm.deployment_group = group
self.schedule("0s", rm.deploy)
if not wait_all_ready:
# schedule a stop. Otherwise the RM will stop immediately
self.schedule("2s", rm.stop_with_conditions)
- def release(self, group = None):
- """ Release the elements of the list 'group' or
- all the resources if any group is specified
+ def release(self, guids = None):
+ """ Release al RMs on the guids list or
+ all the resources if no list is specified
- :param group: List of RM
- :type group: list
+ :param guids: List of RM guids
+ :type guids: list
"""
- if not group:
- group = self.resources
+ if not guids:
+ guids = self.resources
- threads = []
- for guid in group:
+ for guid in guids:
rm = self.get_resource(guid)
- thread = threading.Thread(target=rm.release)
- threads.append(thread)
- thread.setDaemon(True)
- thread.start()
-
- while list(threads) and not self.finished:
- thread = threads[0]
- # Time out after 5 seconds to check EC not terminated
- thread.join(5)
- if not thread.is_alive():
- threads.remove(thread)
+ self.schedule("0s", rm.release)
+
+ self.wait_released(guids)
def shutdown(self):
""" Shutdown the Experiment Controller.
finally:
self.logger.debug("Exiting the task processing loop ... ")
runner.sync()
+ runner.destroy()
def _execute(self, task):
""" Executes a single task.
git://git.onelab.eu / nepi.git / blobdiff