Async setup of TUNs and APPs, for much quicker deployment.

Made dependency setup also more responsive, with an exponential delay that responds better to no-op (which are common since nodes don't get packages uninstalled at cleanup time).

diff --git a/src/nepi/testbeds/planetlab/application.py b/src/nepi/testbeds/planetlab/application.py
index cb39622..c09976f 100644 (file)
--- a/src/nepi/testbeds/planetlab/application.py
+++ b/src/nepi/testbeds/planetlab/application.py
@@ -47,6 +47,8 @@ class Application(object):
         #   Having both pid and ppid makes it harder
         #   for pid rollover to induce tracking mistakes
         self._started = False
+        self._setup = False
+        self._setuper = None
         self._pid = None
         self._ppid = None
     
@@ -212,6 +214,22 @@ class Application(object):
     def setup(self):
         self._make_home()
         self._build()
+        self._setup = True
+    
+    def async_setup(self):
+        if not self._setuper:
+            self._setuper = threading.Thread(
+                target = self.setup)
+            self._setuper.start()
+    
+    def async_setup_wait(self):
+        if not self._setup:
+            if self._setuper:
+                self._setuper.join()
+                if not self._setup:
+                    raise RuntimeError, "Failed to setup application"
+            else:
+                self.setup()
         
     def _make_home(self):
         # Make sure all the paths are created where 

diff --git a/src/nepi/testbeds/planetlab/node.py b/src/nepi/testbeds/planetlab/node.py
index a13f574..db3828b 100644 (file)
--- a/src/nepi/testbeds/planetlab/node.py
+++ b/src/nepi/testbeds/planetlab/node.py
@@ -231,7 +231,7 @@ class Node(object):
             if proc.wait():
                 raise RuntimeError, "Failed to set up application: %s %s" % (out,err,)
     
-    def wait_dependencies(self, pidprobe=1, probe=10, pidmax=10):
+    def wait_dependencies(self, pidprobe=1, probe=0.5, pidmax=10, probemax=10):
         if self.required_packages:
             pidfile = self.DEPENDS_PIDFILE
             
@@ -266,6 +266,7 @@ class Node(object):
                     server_key = self.server_key
                     ):
                 time.sleep(probe)
+                probe = min(probemax, 1.5*probe)
         
     def is_alive(self):
         # Make sure all the paths are created where 

diff --git a/src/nepi/testbeds/planetlab/tunproto.py b/src/nepi/testbeds/planetlab/tunproto.py
index a9f283e..5809b68 100644 (file)
--- a/src/nepi/testbeds/planetlab/tunproto.py
+++ b/src/nepi/testbeds/planetlab/tunproto.py
@@ -6,6 +6,7 @@ import os
 import os.path
 import rspawn
 import subprocess
+import threading
 
 from nepi.util import server
 
@@ -22,6 +23,7 @@ class TunProtoBase(object):
         
         self.home_path = home_path
         
+        self._launcher = None
         self._started = False
         self._pid = None
         self._ppid = None
@@ -91,7 +93,6 @@ class TunProtoBase(object):
         if proc.wait():
             raise RuntimeError, "Failed to set up TUN forwarder: %s %s" % (out,err,)
         
-    
     def launch(self, check_proto, listen, extra_args=[]):
         peer = self.peer()
         local = self.local()
@@ -158,9 +159,25 @@ class TunProtoBase(object):
             )
         
         if proc.wait():
-            raise RuntimeError, "Failed to set up application: %s %s" % (out,err,)
+            raise RuntimeError, "Failed to set up TUN: %s %s" % (out,err,)
 
         self._started = True
+    
+    def async_launch(self, check_proto, listen, extra_args=[]):
+        if not self._launcher:
+            self._launcher = threading.Thread(
+                target = self.launch,
+                args = (check_proto, listen, extra_args))
+            self._launcher.start()
+    
+    def async_launch_wait(self):
+        if not self._started:
+            if self._launcher:
+                self._launcher.join()
+                if not self._started:
+                    raise RuntimeError, "Failed to launch TUN forwarder"
+            else:
+                self.launch()
 
     def checkpid(self):            
         local = self.local()
@@ -295,7 +312,7 @@ class TunProtoUDP(TunProtoBase):
         pass
     
     def setup(self):
-        self.launch('udp', False, ("-U",))
+        self.launch_async('udp', False, ("-U",))
     
     def shutdown(self):
         self.kill()
@@ -307,11 +324,19 @@ class TunProtoTCP(TunProtoBase):
     
     def prepare(self):
         if self.listening:
-            self.launch('tcp', True)
+            self.async_launch('tcp', True)
     
     def setup(self):
         if not self.listening:
+            # make sure our peer is ready
+            peer = self.peer()
+            if peer and peer.peer_proto_impl:
+                peer.peer_proto_impl.async_launch_wait()
+            
             self.launch('tcp', False)
+        else:
+            # make sure WE are ready
+            self.async_launch_wait()
         
         self.checkpid()
     

diff --git a/src/nepi/util/server.py b/src/nepi/util/server.py
index 8c164fd..199a009 100644 (file)
--- a/src/nepi/util/server.py
+++ b/src/nepi/util/server.py
@@ -24,6 +24,7 @@ STOP_MSG = "STOP"
 
 ERROR_LEVEL = 0
 DEBUG_LEVEL = 1
+TRACE = False
 
 if hasattr(os, "devnull"):
     DEV_NULL = os.devnull
@@ -365,6 +366,9 @@ def popen_ssh_command(command, host, port, user, agent,
         """
         Executes a remote commands, returns ((stdout,stderr),process)
         """
+        if TRACE:
+            print "ssh", host, command
+        
         tmp_known_hosts = None
         args = ['ssh',
                 # Don't bother with localhost. Makes test easier
@@ -420,6 +424,9 @@ def popen_scp(source, dest,
         in which case it is advised that the destination be a folder.
         """
         
+        if TRACE:
+            print "scp", source, dest
+        
         if isinstance(source, file) or isinstance(dest, file) \
                 or hasattr(source, 'read')  or hasattr(dest, 'write'):
             assert not recursive