Ticket #45: spanning tree deployment

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

diff --git a/src/nepi/testbeds/planetlab/tunproto.py b/src/nepi/testbeds/planetlab/tunproto.py
index df23a7b..6e518f4 100644 (file)
--- a/src/nepi/testbeds/planetlab/tunproto.py
+++ b/src/nepi/testbeds/planetlab/tunproto.py
@@ -29,6 +29,7 @@ class TunProtoBase(object):
         
         self._launcher = None
         self._started = False
+        self._starting = False
         self._pid = None
         self._ppid = None
         self._if_name = None
@@ -43,7 +44,11 @@ class TunProtoBase(object):
         
         # Make sure all the paths are created where 
         # they have to be created for deployment
-        cmd = "mkdir -p %s" % (server.shell_escape(self.home_path),)
+        # Also remove pidfile, if there is one.
+        # Old pidfiles from previous runs can be troublesome.
+        cmd = "mkdir -p %(home)s ; rm -f %(home)s/pid" % {
+            'home' : server.shell_escape(self.home_path)
+        }
         (out,err),proc = server.popen_ssh_command(
             cmd,
             host = local.node.hostname,
@@ -119,6 +124,11 @@ class TunProtoBase(object):
             raise RuntimeError, "Failed to set up TUN forwarder: %s %s" % (out,err,)
         
     def launch(self, check_proto, listen, extra_args=[]):
+        if self._starting:
+            raise AssertionError, "Double start"
+        
+        self._starting = True
+        
         peer = self.peer()
         local = self.local()
         
@@ -212,6 +222,16 @@ class TunProtoBase(object):
         self._started = True
     
     def _launch_and_wait(self, *p, **kw):
+        try:
+            self.__launch_and_wait(*p, **kw)
+        except:
+            if self._launcher:
+                import sys
+                self._launcher._exc.append(sys.exc_info())
+            else:
+                raise
+            
+    def __launch_and_wait(self, *p, **kw):
         local = self.local()
         
         self.launch(*p, **kw)
@@ -277,13 +297,18 @@ class TunProtoBase(object):
             self._launcher = threading.Thread(
                 target = self._launch_and_wait,
                 args = (check_proto, listen, extra_args))
+            self._launcher._exc = []
             self._launcher.start()
     
     def async_launch_wait(self):
         if self._launcher:
             self._launcher.join()
             if not self._started:
-                raise RuntimeError, "Failed to launch TUN forwarder"
+                if self._launcher._exc:
+                    exctyp,exval,exctrace = self._launcher._exc[0]
+                    raise exctyp,exval,exctrace
+                else:
+                    raise RuntimeError, "Failed to launch TUN forwarder"
         elif not self._started:
             self.launch()
 
@@ -348,8 +373,18 @@ class TunProtoBase(object):
                 agent = None,
                 ident_key = local.node.ident_path,
                 server_key = local.node.server_key,
-                sudo = True
+                sudo = True,
+                nowait = True
                 )
+    
+    def waitkill(self):
+        interval = 1.0
+        for i in xrange(30):
+            status = self.status()
+            if status != rspawn.RUNNING:
+                break
+            time.sleep(interval)
+            interval = min(30.0, interval * 1.1)
         
     def sync_trace(self, local_dir, whichtrace):
         if whichtrace != 'packets':
@@ -409,6 +444,12 @@ class TunProtoBase(object):
         Cleanup
         """
         raise NotImplementedError
+    
+    def destroy(self):
+        """
+        Second-phase cleanup
+        """
+        pass
         
 
 class TunProtoUDP(TunProtoBase):
@@ -425,6 +466,14 @@ class TunProtoUDP(TunProtoBase):
     def shutdown(self):
         self.kill()
 
+    def destroy(self):
+        self.waitkill()
+
+    def launch(self, check_proto='udp', listen=False, extra_args=None):
+        if extra_args is None:
+            extra_args = ("-u",str(self.port))
+        super(TunProtoUDP, self).launch(check_proto, listen, extra_args)
+
 class TunProtoFD(TunProtoBase):
     def __init__(self, local, peer, home_path, key, listening):
         super(TunProtoFD, self).__init__(local, peer, home_path, key)
@@ -439,6 +488,12 @@ class TunProtoFD(TunProtoBase):
     def shutdown(self):
         self.kill()
 
+    def destroy(self):
+        self.waitkill()
+
+    def launch(self, check_proto='fd', listen=False, extra_args=[]):
+        super(TunProtoFD, self).launch(check_proto, listen, extra_args)
+
 class TunProtoTCP(TunProtoBase):
     def __init__(self, local, peer, home_path, key, listening):
         super(TunProtoTCP, self).__init__(local, peer, home_path, key)
@@ -456,16 +511,21 @@ class TunProtoTCP(TunProtoBase):
                 peer.peer_proto_impl.async_launch_wait()
             
             if not self._started:
-                self.launch('tcp', False)
-        else:
-            # make sure WE are ready
-            self.async_launch_wait()
+                self.async_launch('tcp', False)
         
         self.checkpid()
     
     def shutdown(self):
         self.kill()
 
+    def destroy(self):
+        self.waitkill()
+
+    def launch(self, check_proto='tcp', listen=None, extra_args=[]):
+        if listen is None:
+            listen = self.listening
+        super(TunProtoTCP, self).launch(check_proto, listen, extra_args)
+
 class TapProtoUDP(TunProtoUDP):
     def __init__(self, local, peer, home_path, key, listening):
         super(TapProtoUDP, self).__init__(local, peer, home_path, key, listening)