examples/ccnx/planetlab_ccnx_unicast.py

   1 #!/usr/bin/env python
   2
   3 ##
   4 ## Experiment topology:
   5 ##
   6 ##  ccncatchunks                                ccnsendchunks
   7 ##       |                                            |
   8 ##       .->  node1 -- .. -- nodei -- .. -- nodeN   <-.
   9 ##
  10 ##
  11 ##  - Nodes are connected through Intenet
  12 ##  - On each node runs a CCNx daemon
  13 ##  - Static entries are added to the CCNx FIB on each node to communicate them in series.
  14 ##    (Nodes only have FIB entries to at most two nodes)
  15 ##
  16
  17 from nepi.core.design import ExperimentDescription, FactoriesProvider
  18 from nepi.core.execute import ExperimentController
  19 from nepi.util.constants import ApplicationStatus as AS
  20 from optparse import OptionParser, SUPPRESS_HELP
  21 import os
  22 import signal
  23 import string
  24 import subprocess
  25 import tempfile
  26 import time
  27
  28 # Trak SIGTERM, and set global termination flag instead of dying
  29 TERMINATE = []
  30 def _finalize(sig,frame):
  31     global TERMINATE
  32     TERMINATE.append(None)
  33 signal.signal(signal.SIGTERM, _finalize)
  34 signal.signal(signal.SIGINT, _finalize)
  35
  36 def create_slice_desc(slicename, plc_host, pl_user, pl_pwd, pl_ssh_key,
  37         port_base, root_dir, proxy, exp_desc):
  38     pl_provider = FactoriesProvider("planetlab")
  39     slice_desc = exp_desc.add_testbed_description(pl_provider)
  40     slice_desc.set_attribute_value("homeDirectory", root_dir)
  41     slice_desc.set_attribute_value("slice", slicename)
  42     slice_desc.set_attribute_value("sliceSSHKey", pl_ssh_key)
  43     slice_desc.set_attribute_value("authUser", pl_user)
  44     slice_desc.set_attribute_value("authPass", pl_pwd)
  45     slice_desc.set_attribute_value("plcHost", plc_host)
  46     if proxy:
  47         slice_desc.set_attribute_value("proxy", proxy)
  48     slice_desc.set_attribute_value("tapPortBase", port_base)
  49     slice_desc.set_attribute_value("p2pDeployment", True)
  50     # Kills all running processes before starting the experiment
  51     slice_desc.set_attribute_value("cleanProc", True)
  52     # NOTICE: Setting 'cleanHome' to 'True' will erase all previous
  53     # folders in the sliver Home directory, including result files!
  54     #slice_desc.set_attribute_value("cleanHome", True)
  55     slice_desc.set_attribute_value("plLogLevel", "DEBUG")
  56     return slice_desc
  57
  58 def create_node(hostname, pl_inet, slice_desc):
  59     pl_node = slice_desc.create("Node")
  60     pl_node.set_attribute_value("hostname", hostname)
  61     pl_node.set_attribute_value("label", hostname)
  62     pl_iface = slice_desc.create("NodeInterface")
  63     pl_iface.set_attribute_value("label", "iface_%s" % hostname)
  64     pl_iface.connector("inet").connect(pl_inet.connector("devs"))
  65     pl_node.connector("devs").connect(pl_iface.connector("node"))
  66     return pl_node
  67
  68 def create_ccnd(pl_node, port, routes, slice_desc):
  69     pl_app = slice_desc.create("CCNxDaemon")
  70
  71     # We can specify a default ccnx version to be either ccnx-0.5.1 or ccnx-0.6.0
  72     #pl_app.set_attribute_value("ccnxVersion", "ccnx-0.5.1")
  73     # We can also specify a custom local source and build and install directives
  74     path_to_source = os.path.join(os.path.dirname(os.path.abspath(__file__)),
  75         "ccnx-0.6.0rc3.tar.gz")
  76     pl_app.set_attribute_value("sources", path_to_source)
  77     pl_app.set_attribute_value("build",
  78             "tar xzf ${SOURCES}/ccnx-0.6.0rc3.tar.gz && "
  79             "cd ./ccnx-0.6.0rc3 && "
  80             "./configure && make ")
  81     pl_app.set_attribute_value("install", "cp -r ./ccnx-0.6.0rc3/bin ${SOURCES}")
  82
  83     # We use a wildcard to replace the public IP address of the node during runtime,
  84     # once this IP is known
  85     routes = "|".join(map(lambda route: "udp {#[iface_%s].addr[0].[Address]#}" % route, routes))
  86
  87     # Add unicast ccn routes
  88     pl_app.set_attribute_value("ccnRoutes", routes)
  89
  90     # Use a specific port to bind the CCNx daemon
  91     if port:
  92         pl_app.set_attribute_value("ccnLocalPort", port)
  93
  94     pl_app.enable_trace("stdout")
  95     pl_app.enable_trace("stderr")
  96     pl_app.connector("node").connect(pl_node.connector("apps"))
  97
  98 def create_ccnsendchunks(pl_node, port, slice_desc):
  99     pl_app = slice_desc.create("Application")
 100     path_to_video = os.path.join(os.path.dirname(os.path.abspath(__file__)),
 101         "../big_buck_bunny_240p_mpeg4_lq.ts")
 102     pl_app.set_attribute_value("stdin", path_to_video)
 103
 104     command = "ccnsendchunks ccnx:/VIDEO"
 105     if port:
 106         command = "CCN_LOCAL_PORT=%d %s " % (port, command)
 107     pl_app.set_attribute_value("command", command)
 108
 109     pl_app.enable_trace("stdout")
 110     pl_app.enable_trace("stderr")
 111     pl_app.connector("node").connect(pl_node.connector("apps"))
 112     return pl_app
 113
 114 def exec_ccncatchunks(slicename, port, hostname):
 115     print "Getting video chunks from %s ..." % hostname
 116
 117     command = 'PATH=$PATH:$(ls | egrep nepi-ccnd- | head -1)/bin;'
 118     if port:
 119         command += "CCN_LOCAL_PORT=%d " % port
 120     command += ' ccncatchunks2 ccnx:/VIDEO'
 121
 122     login = "%s@%s" % (slicename, hostname)
 123     proc1 = subprocess.Popen(['ssh',
 124         '-o', 'StrictHostKeyChecking=no',
 125         login,
 126         command],
 127         stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell = False)
 128
 129     proc2 = subprocess.Popen(['vlc',
 130         '--ffmpeg-threads=1',
 131         '--sub-filter', 'marq',
 132         '--marq-marquee',
 133         '(c) copyright 2008, Blender Foundation / www.bigbuckbunny.org',
 134         '--marq-position=8',
 135         '--no-video-title-show', '-'],
 136         stdin=proc1.stdout, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
 137     return proc2
 138
 139 def create_ed(hostnames, vsys_vnet, slicename, plc_host, pl_user, pl_pwd, pl_ssh_key,
 140         port_base, root_dir, delay, port, proxy):
 141
 142     # Create the experiment description object
 143     exp_desc = ExperimentDescription()
 144
 145     # Create the slice description object
 146     slice_desc = create_slice_desc(slicename, plc_host, pl_user, pl_pwd, pl_ssh_key,
 147         port_base, root_dir, proxy, exp_desc)
 148
 149     # Create the Internet box object
 150     pl_inet = slice_desc.create("Internet")
 151
 152     # Create the Node boxes
 153     pl_nodes = dict()
 154     ccn_routes = dict()
 155     prev_hostname = None
 156     for hostname in hostnames:
 157         pl_node = create_node(hostname, pl_inet, slice_desc)
 158         pl_nodes[hostname] = pl_node
 159
 160         ccn_routes[hostname] = list()
 161         if prev_hostname:
 162             ccn_routes[hostname].append(prev_hostname)
 163             ccn_routes[prev_hostname].append(hostname)
 164         prev_hostname = hostname
 165
 166     for hostname in hostnames:
 167         pl_node = pl_nodes[hostname]
 168         routes = ccn_routes[hostname]
 169         create_ccnd(pl_node, port, routes, slice_desc)
 170
 171     # Create a ccnsendchunks application box in the first node
 172     hostname = hostnames[0]
 173     pl_node = pl_nodes[hostname]
 174     pl_app = create_ccnsendchunks(pl_node, port, slice_desc)
 175
 176     return exp_desc, pl_nodes, hostname, pl_app
 177
 178 def run(hostnames, vsys_vnet, slicename, plc_host, pl_user, pl_pwd, pl_ssh_key,
 179         port_base, root_dir, delay, port, proxy):
 180
 181     exp_desc, pl_nodes, hostname, pl_app = create_ed(hostnames, vsys_vnet,
 182             slicename, plc_host, pl_user, pl_pwd, pl_ssh_key, port_base,
 183             root_dir, delay, port, proxy)
 184
 185     xml = exp_desc.to_xml()
 186     controller = ExperimentController(xml, root_dir)
 187     controller.start()
 188
 189     while not TERMINATE and controller.status(pl_app.guid) == AS.STATUS_NOT_STARTED:
 190         time.sleep(0.5)
 191
 192     proc1 = None
 193     if not TERMINATE:
 194         hostname = hostnames[-1]
 195         proc1 = exec_ccncatchunks(slicename, port, hostname)
 196
 197     if not TERMINATE and proc1:
 198         time.sleep(delay)
 199
 200     proc2 = None
 201     if not TERMINATE:
 202         hostname = hostnames[-2]
 203         proc2 = exec_ccncatchunks(slicename, port, hostname)
 204
 205     while not TERMINATE and proc1 and proc2 and proc2.poll() is None:
 206         time.sleep(0.5)
 207
 208     if proc1:
 209         if proc1.poll() < 1:
 210            err = proc1.stderr.read()
 211            print "Stream 1 ERROR ", err
 212         else:
 213            out = proc1.stdout.read()
 214            print "Stream 1 OUTPUT ", out
 215
 216     if proc2:
 217         if proc2.poll() < 1:
 218            err = proc2.stderr.read()
 219            print "Stream 2 ERROR ", err
 220         else:
 221            out = proc2.stdout.read()
 222            print "Stream 2 OUTPUT ", out
 223
 224     controller.stop()
 225     controller.shutdown()
 226
 227 if __name__ == '__main__':
 228     root_dir = tempfile.mkdtemp()
 229     slicename = os.environ.get("PL_SLICE")
 230     pl_host = os.environ.get("PL_HOST", "www.planet-lab.eu")
 231     port_base = 2000 + (os.getpid() % 1000) * 13
 232     pl_ssh_key = os.environ.get(
 233         "PL_SSH_KEY",
 234         "%s/.ssh/id_rsa_planetlab" % (os.environ['HOME'],) )
 235     pl_user = os.environ.get('PL_USER')
 236     pl_pwd = os.environ.get('PL_PASS')
 237     pl_vsys_vnet = os.environ.get('PL_VSYS_NET')
 238     pl_hostnames = os.environ.get('PL_HOSTNAMES')
 239     default_hostnames = ['openlab02.pl.sophia.inria.fr',
 240                  'ple4.ipv6.lip6.fr',
 241                  'planetlab2.di.unito.it',
 242                  'merkur.planetlab.haw-hamburg.de',
 243                  'planetlab1.cs.uit.no',
 244                  'planetlab3.cs.st-andrews.ac.uk',
 245                  'planetlab2.cs.uoi.gr',
 246                  'planetlab3.xeno.cl.cam.ac.uk',
 247                  'planet2.inf.tu-dresden.de',
 248                  'planetlab2.csg.uzh.ch',
 249                  'planetlab2.upm.ro',
 250                  'planetlab-um00.di.uminho.pt',
 251                  'planetlabpc2.upf.edu',
 252                  'planet2.elte.hu',
 253                  'planetlab2.esprit-tn.com' ]
 254
 255     ccn_local_port = os.environ.get('CCN_LOCAL_PORT')
 256
 257     usage = "usage: %prog -s <pl_slice> -H <pl_host> -k <ssh_key> -u <pl_user> -p <pl_password> -v <vsys_vnet> -N <host_names> -c <node_count> -d <delay> -P <ccn-local-port> -x <proxy>"
 258
 259     parser = OptionParser(usage=usage)
 260     parser.add_option("-s", "--slicename", dest="slicename",
 261             help="PlanetLab slicename", default=slicename, type="str")
 262     parser.add_option("-H", "--pl-host", dest="pl_host",
 263             help="PlanetLab site (e.g. www.planet-lab.eu)",
 264             default=pl_host, type="str")
 265     parser.add_option("-k", "--ssh-key", dest="pl_ssh_key",
 266             help="Path to private ssh key used for PlanetLab authentication",
 267             default=pl_ssh_key, type="str")
 268     parser.add_option("-u", "--pl-user", dest="pl_user",
 269             help="PlanetLab account user (i.e. Registration email address)",
 270             default=pl_user, type="str")
 271     parser.add_option("-p", "--pl-pwd", dest="pl_pwd",
 272             help="PlanetLab account password", default=pl_pwd, type="str")
 273     parser.add_option("-v", "--vsys-vnet", dest="vsys_vnet",
 274             help="Value of the vsys_vnet tag addigned to your slice. (e.g. 192.168.3.0/16)",
 275             default=pl_vsys_vnet, type="str")
 276     parser.add_option("-N", "--host-names", dest="hostnames",
 277             help="Comma separated list of PlanetLab hostnames to use",
 278             default=pl_hostnames, type="str")
 279     parser.add_option("-c", "--node-count", dest="node_count",
 280             help="Number of nodes to use",
 281             default=9, type="int")
 282     parser.add_option("-d", "--delay", dest="delay",
 283             help="Time to wait before retrieveing the second video stream in seconds",
 284             default=40, type="int")
 285     parser.add_option("-P", "--ccn-local-port", dest="port",
 286             help="Port to bind the CCNx daemon",
 287             default=ccn_local_port, type="int")
 288     parser.add_option("-x", "--proxy", dest="proxy",
 289             help="Https proxy between here and PlanetLab machines",
 290             default=None, type="str")
 291     (options, args) = parser.parse_args()
 292
 293     hostnames = map(string.strip, options.hostnames.split(",")) if options.hostnames else default_hostnames
 294     if options.node_count > 0 and options.node_count < len(hostnames):
 295        hostnames = hostnames[0:options.node_count]
 296
 297     vsys_vnet = options.vsys_vnet
 298     slicename = options.slicename
 299     pl_host = options.pl_host
 300     pl_user= options.pl_user
 301     pl_pwd = options.pl_pwd
 302     pl_ssh_key = options.pl_ssh_key
 303     delay = options.delay
 304     port = options.port
 305     proxy = options.proxy
 306
 307     run(hostnames, vsys_vnet, slicename, pl_host, pl_user, pl_pwd, pl_ssh_key,
 308             port_base, root_dir, delay, port, proxy)
 309