Make PlanetLab select lightly loaded nodes when given the chance (ie, when more candi...

[nepi.git] / src / nepi / testbeds / planetlab / execute.py

diff --git a/src/nepi/testbeds/planetlab/execute.py b/src/nepi/testbeds/planetlab/execute.py
index 292b92e..dcb5023 100644 (file)
--- a/src/nepi/testbeds/planetlab/execute.py
+++ b/src/nepi/testbeds/planetlab/execute.py
@@ -37,10 +37,11 @@ class TestbedController(testbed_impl.TestbedController):
         self.slicename = None
         self._traces = dict()
 
-        import node, interfaces, application
+        import node, interfaces, application, multicast
         self._node = node
         self._interfaces = interfaces
         self._app = application
+        self._multicast = multicast
         
         self._blacklist = set()
         self._just_provisioned = set()
@@ -155,6 +156,10 @@ class TestbedController(testbed_impl.TestbedController):
             raise RuntimeError, "PlanetLab account username not set"
         if not self.authString:
             raise RuntimeError, "PlanetLab account passphrase not set"
+        if not self.sliceSSHKey:
+            raise RuntimeError, "PlanetLab account key not specified"
+        if not os.path.exists(self.sliceSSHKey):
+            raise RuntimeError, "PlanetLab account key cannot be opened: %s" % (self.sliceSSHKey,)
         
         self._logger.setLevel(getattr(logging,self.logLevel))
         
@@ -251,9 +256,18 @@ class TestbedController(testbed_impl.TestbedController):
         if nodes and reqs:
             if recover:
                 raise RuntimeError, "Impossible to recover: unassigned host for Nodes %r" % (nodes,)
+
+            def pickbest(fullset, nreq, node=nodes[0]):
+                if len(fullset) > nreq:
+                    fullset = zip(node.rate_nodes(fullset),fullset)
+                    fullset.sort(reverse=True)
+                    del fullset[nreq:]
+                    return set(map(operator.itemgetter(1),fullset))
+                else:
+                    return fullset
             
             try:
-                solution = resourcealloc.alloc(reqs)
+                solution = resourcealloc.alloc(reqs, sample=pickbest)
             except resourcealloc.ResourceAllocationError:
                 # Failed, try again with all nodes
                 reqs = []
@@ -262,7 +276,7 @@ class TestbedController(testbed_impl.TestbedController):
                     candidates -= reserved
                     reqs.append(candidates)
                 
-                solution = resourcealloc.alloc(reqs)
+                solution = resourcealloc.alloc(reqs, sample=pickbest)
                 to_provision.update(solution)
             
             # Do assign nodes
@@ -544,10 +558,10 @@ class TestbedController(testbed_impl.TestbedController):
         self._traces.clear()
 
     def trace(self, guid, trace_id, attribute='value'):
-        app = self._elements[guid]
+        elem = self._elements[guid]
 
         if attribute == 'value':
-            path = app.sync_trace(self.home_directory, trace_id)
+            path = elem.sync_trace(self.home_directory, trace_id)
             if path:
                 fd = open(path, "r")
                 content = fd.read()
@@ -555,9 +569,9 @@ class TestbedController(testbed_impl.TestbedController):
             else:
                 content = None
         elif attribute == 'path':
-            content = app.remote_trace_path(trace_id)
+            content = elem.remote_trace_path(trace_id)
         elif attribute == 'name':
-            content = app.remote_trace_name(trace_id)
+            content = elem.remote_trace_name(trace_id)
         else:
             content = None
         return content
@@ -708,3 +722,19 @@ class TestbedController(testbed_impl.TestbedController):
     def _make_tun_filter(self, parameters):
         return self._make_generic(parameters, self._interfaces.TunFilter)
 
+    def _make_class_queue_filter(self, parameters):
+        return self._make_generic(parameters, self._interfaces.ClassQueueFilter)
+
+    def _make_tos_queue_filter(self, parameters):
+        return self._make_generic(parameters, self._interfaces.ToSQueueFilter)
+
+    def _make_multicast_forwarder(self, parameters):
+        return self._make_generic(parameters, self._multicast.MulticastForwarder)
+
+    def _make_multicast_announcer(self, parameters):
+        return self._make_generic(parameters, self._multicast.MulticastAnnouncer)
+
+    def _make_multicast_router(self, parameters):
+        return self._make_generic(parameters, self._multicast.MulticastRouter)
+
+