From 244c895c8fa1ce97f21db9f6d2d7274c7b063995 Mon Sep 17 00:00:00 2001
From: Claudio-Daniel Freire <claudio-daniel.freire@inria.fr>
Date: Fri, 29 Apr 2011 13:06:47 +0200
Subject: [PATCH] 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).
---
src/nepi/testbeds/planetlab/application.py | 18 ++++++++++++
src/nepi/testbeds/planetlab/node.py | 3 +-
src/nepi/testbeds/planetlab/tunproto.py | 33 +++++++++++++++++++---
src/nepi/util/server.py | 7 +++++
4 files changed, 56 insertions(+), 5 deletions(-)
diff --git a/src/nepi/testbeds/planetlab/application.py b/src/nepi/testbeds/planetlab/application.py
index cb396222..c09976f7 100644
--- 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 a13f5741..db3828be 100644
--- 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 a9f283e8..5809b684 100644
--- 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 8c164fd5..199a0095 100644
--- 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
--
2.47.0