from nepi.util import guid
from nepi.util.parallel import ParallelRun
-from nepi.util.timefuncs import tnow, tdiffsec, stabsformat
+from nepi.util.timefuncs import tnow, tdiffsec, stabsformat, tsformat
from nepi.execution.resource import ResourceFactory, ResourceAction, \
ResourceState, ResourceState2str
from nepi.execution.scheduler import HeapScheduler, Task, TaskStatus
from nepi.execution.trace import TraceAttr
# TODO: use multiprocessing instead of threading
-# TODO: When a failure occurrs during deployment scp and ssh processes are left running behind!!
+# 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)
class ECState(object):
""" State of the Experiment Controller
"""
.. class:: Class Args :
- :param exp_id: Human readable identifier for the experiment.
- It will be used in the name of the directory
+ :param exp_id: Human readable identifier for the experiment scenario.
+ It will be used in the name of the directory
where experiment related information is stored
- :type exp_id: int
-
- :param root_dir: Root directory where experiment specific folder
- will be created to store experiment information
- :type root_dir: str
+ :type exp_id: str
.. note::
+
+ An experiment, or scenario, is defined by a concrete use, behavior,
+ configuration and interconnection of resources that describe a single
+ experiment case (We call this the experiment description).
+ A same experiment (scenario) can be run many times.
+
The ExperimentController (EC), is the entity responsible for
- managing a single experiment.
+ managing an experiment instance (run). The same scenario can be
+ recreated (and re-run) by instantiating an EC and recreating
+ the same experiment description.
+
+ In NEPI, an experiment is represented as a graph of interconnected
+ resources. A resource is a generic concept in the sense that any
+ component taking part of an experiment, whether physical of
+ virtual, is considered a resource. A resources could be a host,
+ a virtual machine, an application, a simulator, a IP address.
+
+ A ResourceManager (RM), is the entity responsible for managing a
+ single resource. ResourceManagers are specific to a resource
+ type (i.e. An RM to control a Linux application will not be
+ the same as the RM used to control a ns-3 simulation).
+ In order for a new type of resource to be supported in NEPI
+ a new RM must be implemented. NEPI already provides different
+ RMs to control basic resources, and new can be extended from
+ the existing ones.
+
Through the EC interface the user can create ResourceManagers (RMs),
- configure them and interconnect them, in order to describe the experiment.
-
- Only when the 'deploy()' method is invoked, the EC will take actions
- to transform the 'described' experiment into a 'running' experiment.
+ configure them and interconnect them, in order to describe an experiment.
+ Describing an experiment through the EC does not run the experiment.
+ Only when the 'deploy()' method is invoked on the EC, will the EC take
+ actions to transform the 'described' experiment into a 'running' experiment.
While the experiment is running, it is possible to continue to
create/configure/connect RMs, and to deploy them to involve new
- resources in the experiment.
-
+ resources in the experiment (this is known as 'interactive' deployment).
+
+ An experiments in NEPI is identified by a string id,
+ which is either given by the user, or automatically generated by NEPI.
+ The purpose of this identifier is to separate files and results that
+ belong to different experiment scenarios.
+ However, since a same 'experiment' can be run many times, the experiment
+ id is not enough to identify an experiment instance (run).
+ For this reason, the ExperimentController has two identifier, the
+ exp_id, which can be re-used by different ExperimentController instances,
+ and the run_id, which unique to a ExperimentController instance, and
+ is automatically generated by NEPI.
+
"""
- def __init__(self, exp_id = None, root_dir = "/tmp"):
+ def __init__(self, exp_id = None):
super(ExperimentController, self).__init__()
# root directory to store files
- self._root_dir = root_dir
- # experiment identifier given by the user
- self._exp_id = exp_id or "nepi-exp-%s" % os.urandom(8).encode('hex')
+ # Run identifier. It identifies a concrete instance (run) of an experiment.
+ # Since a same experiment (same configuration) can be run many times,
+ # this id permits to identify concrete exoeriment run
+ self._run_id = tsformat()
+
+ # Experiment identifier. Usually assigned by the user
+ self._exp_id = exp_id or "exp-%s" % os.urandom(8).encode('hex')
# generator of globally unique ids
self._guid_generator = guid.GuidGenerator()
@property
def exp_id(self):
- """ Return the experiment ID
+ """ Return the experiment id assigned by the user
"""
- exp_id = self._exp_id
- if not exp_id.startswith("nepi-"):
- exp_id = "nepi-" + exp_id
- return exp_id
+ return self._exp_id
+
+ @property
+ def run_id(self):
+ """ Return the experiment instance (run) identifier
+
+ """
+ return self._run_id
@property
def finished(self):
"""
return self.wait(guids, states = [ResourceState.STARTED,
ResourceState.STOPPED,
+ ResourceState.FAILED,
ResourceState.FINISHED])
def wait(self, guids, states = [ResourceState.FINISHED,
- ResourceState.STOPPED]):
+ ResourceState.FAILED,
+ 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]
+ state = self.state(guid)
+
+ if state in states:
+ guids.remove(guid)
+ else:
+ # Debug...
+ self.logger.debug(" WAITING FOR %g - state %s " % (guid,
+ self.state(guid, hr = True)))
+
+ # 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(2)
+
def get_task(self, tid):
""" Get a specific task
git://git.onelab.eu / nepi.git / blobdiff