Supporting many concurrent LinuxApplications on same LinuxNode

[nepi.git] / src / neco / execution / ec.py

diff --git a/src/neco/execution/ec.py b/src/neco/execution/ec.py
index 1b7d4be..c9e4e06 100644 (file)
--- a/src/neco/execution/ec.py
+++ b/src/neco/execution/ec.py
@@ -5,26 +5,33 @@ import time
 import threading
 
 from neco.util import guid
+from neco.util.parallel import ParallelRun
 from neco.util.timefuncs import strfnow, strfdiff, strfvalid 
 from neco.execution.resource import ResourceFactory, ResourceAction, \
         ResourceState
 from neco.execution.scheduler import HeapScheduler, Task, TaskStatus
-from neco.util.parallel import ParallelRun
+from neco.execution.trace import TraceAttr
 
 # TODO: use multiprocessing instead of threading
 
 class ExperimentController(object):
-    def __init__(self, root_dir = "/tmp", loglevel = 'error'): 
+    def __init__(self, exp_id = None, root_dir = "/tmp"): 
         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')
+
         # generator of globally unique ids
         self._guid_generator = guid.GuidGenerator()
         
         # Resource managers
         self._resources = dict()
 
+        # Resource managers
+        self._group = dict()
+
         # Scheduler
         self._scheduler = HeapScheduler()
 
@@ -38,8 +45,18 @@ class ExperimentController(object):
         self._thread.start()
 
         # Logging
-        self._logger = logging.getLogger("neco.execution.ec")
-        self._logger.setLevel(getattr(logging, loglevel.upper()))
+        self._logger = logging.getLogger("ExperimentController")
+
+    @property
+    def logger(self):
+        return self._logger
+
+    @property
+    def exp_id(self):
+        exp_id = self._exp_id
+        if not exp_id.startswith("nepi-"):
+            exp_id = "nepi-" + exp_id
+        return exp_id
 
     def get_task(self, tid):
         return self._tasks.get(tid)
@@ -51,18 +68,28 @@ class ExperimentController(object):
     def resources(self):
         return self._resources.keys()
 
-    def register_resource(self, rtype, guid = None, creds = None):
+    def register_resource(self, rtype, guid = None):
         # Get next available guid
         guid = self._guid_generator.next(guid)
         
         # Instantiate RM
-        rm = ResourceFactory.create(rtype, self, guid, creds)
+        rm = ResourceFactory.create(rtype, self, guid)
 
         # Store RM
         self._resources[guid] = rm
 
         return guid
 
+    def register_group(self, group):
+        guid = self._guid_generator.next()
+
+        if not isinstance(group, list):
+            group = [group] 
+
+        self._groups[guid] = group
+
+        return guid
+
     def get_attributes(self, guid):
         rm = self.get_resource(guid)
         return rm.get_attributes()
@@ -100,14 +127,47 @@ class ExperimentController(object):
 
         """
         if isinstance(group1, int):
-            group1 = list[group1]
+            group1 = [group1]
         if isinstance(group2, int):
-            group2 = list[group2]
+            group2 = [group2]
 
         for guid1 in group1:
-            rm = self.get_resource(guid)
+            rm = self.get_resource(guid1)
             rm.register_condition(action, group2, state, time)
 
+    def register_trace(self, guid, name):
+        """ Enable trace
+
+        :param name: Name of the trace
+        :type name: str
+        """
+        rm = self.get_resource(guid)
+        rm.register_trace(name)
+
+    def trace(self, guid, name, attr = TraceAttr.ALL, block = 512, offset = 0):
+        """ Get information on collected trace
+
+        :param name: Name of the trace
+        :type name: str
+
+        :param attr: Can be one of:
+                         - TraceAttr.ALL (complete trace content), 
+                         - TraceAttr.STREAM (block in bytes to read starting at offset), 
+                         - TraceAttr.PATH (full path to the trace file),
+                         - TraceAttr.SIZE (size of trace file). 
+        :type attr: str
+
+        :param block: Number of bytes to retrieve from trace, when attr is TraceAttr.STREAM 
+        :type name: int
+
+        :param offset: Number of 'blocks' to skip, when attr is TraceAttr.STREAM 
+        :type name: int
+
+        :rtype: str
+        """
+        rm = self.get_resource(guid)
+        return rm.trace(name, attr, block, offset)
+
     def discover(self, guid, filters):
         rm = self.get_resource(guid)
         return rm.discover(filters)
@@ -165,9 +225,9 @@ class ExperimentController(object):
 
         """
         if isinstance(group1, int):
-            group1 = list[group1]
+            group1 = [group1]
         if isinstance(group2, int):
-            group2 = list[group2]
+            group2 = [group2]
 
         for guid1 in group1:
             rm = self.get_resource(guid)
@@ -187,11 +247,13 @@ class ExperimentController(object):
         :param group: List of guids of RMs to deploy
         :type group: list
 
-        :param wait_all_ready: Wait until all RMs are deployed in
+        :param wait_all_ready: Wait until all RMs are ready in
             order to start the RMs
         :type guid: int
 
         """
+        self.logger.debug(" ------- DEPLOY START ------ ")
+
         def steps(rm):
             rm.deploy()
             rm.start_with_conditions()
@@ -212,9 +274,9 @@ class ExperimentController(object):
                 towait = list(group)
                 towait.remove(guid)
                 self.register_condition(guid, ResourceAction.START, 
-                        towait, ResourceState.DEPLOYED)
+                        towait, ResourceState.READY)
 
-            thread = threading.Thread(target = steps, args = (rm))
+            thread = threading.Thread(target = steps, args = (rm,))
             threads.append(thread)
             thread.start()