From: Alina Quereilhac Date: Fri, 7 Feb 2014 18:53:35 +0000 (+0100) Subject: Trying to make LinuxNS3Simulator to deploy remotely .... X-Git-Tag: nepi-3.1.0~120^2~11 X-Git-Url: http://git.onelab.eu/?a=commitdiff_plain;h=68adac66099b08e3daae7a84b29af0f7c69ee955;p=nepi.git Trying to make LinuxNS3Simulator to deploy remotely .... --- diff --git a/setup.py b/setup.py index e618057f..01ac52aa 100755 --- a/setup.py +++ b/setup.py @@ -31,6 +31,7 @@ setup( package_dir = {"": "src"}, package_data = { "nepi.resources.planetlab" : [ "scripts/*.py" ], - "nepi.resources.linux" : [ "scripts/*.py" ] + "nepi.resources.linux" : [ "scripts/*.py" ], + "nepi.resources.linux.ns3" : [ "dependencies/*.tar.gz" ] } ) diff --git a/src/nepi/execution/ec.py b/src/nepi/execution/ec.py index 571b3d5c..7d8b990b 100644 --- a/src/nepi/execution/ec.py +++ b/src/nepi/execution/ec.py @@ -31,7 +31,6 @@ from nepi.execution.trace import TraceAttr import functools import logging import os -import random import sys import time import threading diff --git a/src/nepi/resources/linux/application.py b/src/nepi/resources/linux/application.py index e287a8af..642b287c 100644 --- a/src/nepi/resources/linux/application.py +++ b/src/nepi/resources/linux/application.py @@ -507,10 +507,6 @@ class LinuxApplication(ResourceManager): x11 = self.get("forwardX11") env = self.get("env") - # For a command being executed in foreground, if there is stdin, - # it is expected to be text string not a file or pipe - stdin = self.get("stdin") or None - # Command will be launched in foreground and attached to the # terminal using the node 'execute' in non blocking mode. @@ -521,7 +517,6 @@ class LinuxApplication(ResourceManager): (out, err), self._proc = self.execute_command(command, env = env, sudo = sudo, - stdin = stdin, forward_x11 = x11, blocking = False) @@ -669,7 +664,6 @@ class LinuxApplication(ResourceManager): def execute_command(self, command, env = None, sudo = False, - stdin = None, forward_x11 = False, blocking = False): @@ -681,7 +675,6 @@ class LinuxApplication(ResourceManager): return self.node.execute(command, sudo = sudo, - stdin = stdin, forward_x11 = forward_x11, blocking = blocking) diff --git a/src/nepi/resources/linux/node.py b/src/nepi/resources/linux/node.py index ee6475ce..edfa0cae 100644 --- a/src/nepi/resources/linux/node.py +++ b/src/nepi/resources/linux/node.py @@ -422,9 +422,40 @@ class LinuxNode(ResourceManager): return self.execute(cmd, with_lock = True) + def socat(self, local_socket_name, + remote_socket_name, + sudo = False, + identity = None, + server_key = None, + env = None, + tty = False, + connect_timeout = 30, + retry = 3, + strict_host_checking = True): + """ Connectes a local and a remote UNIX socket through SSH using socat """ + + if self.localhost: + return (None, None), None + else: + return sshfuncs.socat( + local_socket_name, + remote_socket_name, + host = self.get("hostname"), + user = self.get("username"), + port = self.get("port"), + agent = True, + sudo = sudo, + identity = self.get("identity"), + server_key = self.get("serverKey"), + env = env, + tty = tty, + retry = retry, + connect_timeout = connect_timeout, + strict_host_checking = strict_host_checking + ) + def execute(self, command, sudo = False, - stdin = None, env = None, tty = False, forward_x11 = False, @@ -443,10 +474,13 @@ class LinuxNode(ResourceManager): (out, err), proc = execfuncs.lexec(command, user = self.get("username"), # still problem with localhost sudo = sudo, - stdin = stdin, env = env) else: if with_lock: + # If the execute command is blocking, we don't want to keep + # the node lock. This lock is used to avoid race conditions + # when creating the ControlMaster sockets. A more elegant + # solution is needed. with self._node_lock: (out, err), proc = sshfuncs.rexec( command, @@ -455,7 +489,6 @@ class LinuxNode(ResourceManager): port = self.get("port"), agent = True, sudo = sudo, - stdin = stdin, identity = self.get("identity"), server_key = self.get("serverKey"), env = env, @@ -475,7 +508,6 @@ class LinuxNode(ResourceManager): port = self.get("port"), agent = True, sudo = sudo, - stdin = stdin, identity = self.get("identity"), server_key = self.get("serverKey"), env = env, @@ -595,14 +627,13 @@ class LinuxNode(ResourceManager): recursive = True, strict_host_checking = False) else: - with self._node_lock: - (out, err), proc = sshfuncs.rcopy( - src, dst, - port = self.get("port"), - identity = self.get("identity"), - server_key = self.get("serverKey"), - recursive = True, - strict_host_checking = False) + (out, err), proc = sshfuncs.rcopy( + src, dst, + port = self.get("port"), + identity = self.get("identity"), + server_key = self.get("serverKey"), + recursive = True, + strict_host_checking = False) return (out, err), proc diff --git a/src/nepi/resources/linux/ns3/dependencies/pygccxml-1.0.0.tar.gz b/src/nepi/resources/linux/ns3/dependencies/pygccxml-1.0.0.tar.gz new file mode 100644 index 00000000..5e71694f Binary files /dev/null and b/src/nepi/resources/linux/ns3/dependencies/pygccxml-1.0.0.tar.gz differ diff --git a/src/nepi/resources/linux/ns3/ns3client.py b/src/nepi/resources/linux/ns3/ns3client.py index 3fe5b99c..59b04589 100644 --- a/src/nepi/resources/linux/ns3/ns3client.py +++ b/src/nepi/resources/linux/ns3/ns3client.py @@ -21,25 +21,38 @@ import base64 import cPickle import errno import socket +import weakref + from optparse import OptionParser, SUPPRESS_HELP from nepi.resources.ns3.ns3client import NS3Client from nepi.resources.ns3.ns3server import NS3WrapperMessage class LinuxNS3Client(NS3Client): - def __init__(self, socket_name): + def __init__(self, simulation): super(LinuxNS3Client, self).__init__() - self._socket_name = socket_name + self._simulation = weakref.ref(simulation) + + self._socat_proc = None + self.connect_client() @property - def socket_name(self): - return self._socket_name + def simulation(self): + return self._simulation() + + def connect_client(self): + if self.simulation.node.get("hostname") in ['localhost', '127.0.0.1']: + return + + (out, err), self._socat_proc = self.simulation.node.socat( + self.simulation.local_socket, + self.simulation.remote_socket) def send_msg(self, msg, *args): args = list(args) sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) - sock.connect(self.socket_name) + sock.connect(self.simulation.local_socket) args.insert(0, msg) def encode(arg): @@ -95,5 +108,23 @@ class LinuxNS3Client(NS3Client): return self.send_msg(NS3WrapperMessage.STOP, *args) def shutdown(self): - return self.send_msg(NS3WrapperMessage.SHUTDOWN, []) + ret = None + + try: + ret = self.send_msg(NS3WrapperMessage.SHUTDOWN, []) + except: + pass + + try: + if self._socat_proc: + self._socat_proc.kill() + except: + pass + + try: + os.remove(self.simulation.local_socket) + except: + pass + + return ret diff --git a/src/nepi/resources/linux/ns3/ns3simulation.py b/src/nepi/resources/linux/ns3/ns3simulation.py new file mode 100644 index 00000000..e3d41f7a --- /dev/null +++ b/src/nepi/resources/linux/ns3/ns3simulation.py @@ -0,0 +1,320 @@ +# +# NEPI, a framework to manage network experiments +# Copyright (C) 2014 INRIA +# +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with this program. If not, see . +# +# Author: Alina Quereilhac + +from nepi.execution.attribute import Attribute, Flags, Types +from nepi.execution.trace import Trace, TraceAttr +from nepi.execution.resource import ResourceManager, clsinit_copy, \ + ResourceState, reschedule_delay +from nepi.resources.linux.application import LinuxApplication +from nepi.util.timefuncs import tnow, tdiffsec +from nepi.resources.ns3.ns3simulation import NS3Simulation +from nepi.resources.linux.ns3.ns3client import LinuxNS3Client + +import os + +@clsinit_copy +class LinuxNS3Simulation(LinuxApplication, NS3Simulation): + _rtype = "LinuxNS3Simulation" + + @classmethod + def _register_attributes(cls): + ns_log = Attribute("nsLog", + "NS_LOG environment variable ", + flags = Flags.Design) + + verbose = Attribute("verbose", + "True to output debugging info from the ns3 client-server communication", + type = Types.Bool, + flags = Flags.Design) + + cls._register_attribute(ns_log) + + cls._register_attribute(verbose) + + def __init__(self, ec, guid): + LinuxApplication.__init__(self, ec, guid) + NS3Simulation.__init__(self) + + self._client = None + self._home = "ns3-simu-%s" % self.guid + self._socket_name = "ns3simu-%s" % os.urandom(8).encode('hex') + + @property + def socket_name(self): + return self._socket_name + + @property + def remote_socket(self): + return os.path.join(self.run_home, self.socket_name) + + @property + def local_socket(self): + if self.node.get('hostname') in ['localhost', '127.0.0.01']: + return self.remote_socket + + return os.path.join("/", "tmp", self.socket_name) + + def upload_sources(self): + self.node.mkdir(os.path.join(self.node.src_dir, "ns3wrapper")) + + # upload ns3 wrapper python script + ns3_wrapper = os.path.join(os.path.dirname(__file__), "..", "..", "ns3", + "ns3wrapper.py") + + self.node.upload(ns3_wrapper, + os.path.join(self.node.src_dir, "ns3wrapper", "ns3wrapper.py"), + overwrite = False) + + # upload ns3_server python script + ns3_server = os.path.join(os.path.dirname(__file__), "..", "..", "ns3", + "ns3server.py") + + self.node.upload(ns3_server, + os.path.join(self.node.src_dir, "ns3wrapper", "ns3server.py"), + overwrite = False) + + if self.node.use_rpm: + # upload pygccxml sources + pygccxml_tar = os.path.join(os.path.dirname(__file__), "dependencies", + "%s.tar.gz" % self.pygccxml_version) + + self.node.upload(pygccxml_tar, + os.path.join(self.node.src_dir, "%s.tar.gz" % self.pygccxml_version), + overwrite = False) + + def upload_start_command(self): + command = self.get("command") + env = self.get("env") + + # We want to make sure the ccnd is running + # before the experiment starts. + # Run the command as a bash script in background, + # in the host ( but wait until the command has + # finished to continue ) + env = self.replace_paths(env) + command = self.replace_paths(command) + + shfile = os.path.join(self.app_home, "start.sh") + self.node.upload_command(command, + shfile = shfile, + env = env, + overwrite = True) + + # Run the ns3wrapper + self._run_in_background() + + def do_deploy(self): + if not self.node or self.node.state < ResourceState.READY: + self.debug("---- RESCHEDULING DEPLOY ---- node state %s " % self.node.state ) + + # ccnd needs to wait until node is deployed and running + self.ec.schedule(reschedule_delay, self.deploy) + else: + if not self.get("command"): + self.set("command", self._start_command) + + if not self.get("depends"): + self.set("depends", self._dependencies) + + if not self.get("build"): + self.set("build", self._build) + + if not self.get("install"): + self.set("install", self._install) + + if not self.get("env"): + self.set("env", self._environment) + + self.do_discover() + self.do_provision() + + # Create client + self._client = LinuxNS3Client(self) + + self.set_ready() + + def do_start(self): + """ Starts simulation execution + + """ + self.info("Starting ns-3 simulation") + + if self.state == ResourceState.READY: + self.set_started() + else: + msg = " Failed to execute command '%s'" % command + self.error(msg, out, err) + raise RuntimeError, msg + + def do_stop(self): + """ Stops simulation execution + + """ + if self.state == ResourceState.STARTED: + # TODO: Stop simulation + LinuxApplication.do_stop(self) + + def do_release(self): + self.info("Releasing resource") + + tear_down = self.get("tearDown") + if tear_down: + self.node.execute(tear_down) + + self.do_stop() + + self._client.shutdown() + + super(LinuxApplication, self).do_release() + + @property + def _start_command(self): + command = [] + command.append("PYTHONPATH=$PYTHONPATH:${SRC}/ns3wrapper/") + + command.append("python ${SRC}/ns3wrapper/ns3server.py -S %s" % self.remote_socket ) + + ns_log = self.get("nsLog") + if ns_log: + command.append("-L %s" % ns_log) + if self.get("verbose"): + command.append("-v") + + command.append("-H") + command.append(self.run_home) + + command = " ".join(command) + return command + + @property + def _dependencies(self): + if self.node.use_rpm: + return ( " gcc gcc-c++ python python-devel mercurial bzr tcpdump socat gccxml") + elif self.node.use_deb: + return ( " gcc g++ python python-dev mercurial bzr tcpdump socat gccxml python-pygccxml") + return "" + + @property + def ns3_repo(self): + return "http://code.nsnam.org" + + @property + def ns3_version(self): + return "ns-3.19" + + @property + def pybindgen_version(self): + return "834" + + @property + def pygccxml_version(self): + return "pygccxml-1.0.0" + + @property + def _build(self): + return ( + # Test if ns-3 is alredy installed + " ( " + " (( " + " ( test -d ${SRC}/%(ns3_version)s ) || (test -d ${NS3BINDINGS:='None'} && test -d ${NS3LIBRARIES:='None'}) ) && " + " echo 'binaries found, nothing to do' )" + " ) " + " || " + # If not, install ns-3 and its dependencies + " ( " + # Install pygccxml + " ( " + " ( " + " python -c 'import pygccxml' && " + " echo 'pygccxml not found' " + " ) " + " || " + " ( " + " tar xf ${SRC}/%(pygccxml_version)s.tar.gz -C ${SRC} && " + " cd ${SRC}/%(pygccxml_version)s && " + " sudo -S python setup.py install " + " ) " + " ) " + # Install pybindgen + " && " + " ( " + " ( " + " test -d ${BIN}/pybindgen && " + " echo 'binaries found, nothing to do' " + " ) " + " || " + # If not, clone and build + " ( cd ${SRC} && " + " bzr checkout lp:pybindgen -r %(pybindgen_version)s && " + " cd ${SRC}/pybindgen && " + " ./waf configure && " + " ./waf " + " ) " + " ) " + " && " + # Clone and build ns-3 + " ( " + " hg clone %(ns3_repo)s/%(ns3_version)s ${SRC}/%(ns3_version)s && " + " cd ${SRC}/%(ns3_version)s && " + " ./waf configure -d optimized && " + " ./waf " + " ) " + " ) " + ) % ({ + 'ns3_repo': self.ns3_repo, + 'ns3_version': self.ns3_version, + 'pybindgen_version': self.pybindgen_version, + 'pygccxml_version': self.pygccxml_version + }) + + @property + def _install(self): + return ( + # Test if ns-3 is alredy cloned + " ( " + " ( ( (test -d ${BIN}/%(ns3_version)s/build ) || " + " (test -d ${NS3BINDINGS:='None'} && test -d ${NS3LIBRARIES:='None'}) ) && " + " echo 'binaries found, nothing to do' )" + " ) " + " ||" + " ( " + # If not, copy ns-3 build to bin + " mkdir -p ${BIN}/%(ns3_version)s && " + " mv ${SRC}/%(ns3_version)s/build ${BIN}/%(ns3_version)s/build " + " )" + ) % ({ + 'ns3_version': self.ns3_version + }) + + @property + def _environment(self): + env = [] + env.append("NS3BINDINGS=${NS3BINDINGS:=${BIN}/%(ns3_version)s/build/bindings/python/}" % ({ + 'ns3_version': self.ns3_version, + })) + env.append("NS3LIBRARIES=${NS3LIBRARIES:=${BIN}/%(ns3_version)s/build/}" % ({ + 'ns3_version': self.ns3_version, + })) + + return " ".join(env) + + def valid_connection(self, guid): + # TODO: Validate! + return True + diff --git a/src/nepi/resources/linux/ns3/ns3simulator.py b/src/nepi/resources/linux/ns3/ns3simulator.py deleted file mode 100644 index e468005f..00000000 --- a/src/nepi/resources/linux/ns3/ns3simulator.py +++ /dev/null @@ -1,75 +0,0 @@ -# -# NEPI, a framework to manage network experiments -# Copyright (C) 2014 INRIA -# -# This program is free software: you can redistribute it and/or modify -# it under the terms of the GNU General Public License as published by -# the Free Software Foundation, either version 3 of the License, or -# (at your option) any later version. -# -# This program is distributed in the hope that it will be useful, -# but WITHOUT ANY WARRANTY; without even the implied warranty of -# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -# GNU General Public License for more details. -# -# You should have received a copy of the GNU General Public License -# along with this program. If not, see . -# -# Author: Alina Quereilhac - -from nepi.execution.attribute import Attribute, Flags, Types -from nepi.execution.trace import Trace, TraceAttr -from nepi.execution.resource import ResourceManager, clsinit_copy, \ - ResourceState, reschedule_delay -from nepi.resources.linux.application import LinuxApplication -from nepi.util.timefuncs import tnow, tdiffsec -from nepi.resources.ns3.ns3simulator import NS3Simulator -from nepi.resources.linux.ns3.ns3client import LinuxNS3Client - -import os - -@clsinit_copy -class LinuxNS3Simulator(LinuxApplication, NS3Simulator): - _rtype = "LinuxNS3Simulator" - - @classmethod - def _register_attributes(cls): - socket_name = Attribute("socketName", - "Local socket name to communicate with the ns-3 server ", - flags = Flags.Design) - - cls._register_attribute(socket_name) - - def __init__(self, ec, guid): - super(LinuxApplication, self).__init__(ec, guid) - super(NS3Simulator, self).__init__() - - self._client = None - self._home = "ns3-simu-%s" % self.guid - - def do_deploy(self): - if not self.node or self.node.state < ResourceState.READY: - self.debug("---- RESCHEDULING DEPLOY ---- node state %s " % self.node.state ) - - # ccnd needs to wait until node is deployed and running - self.ec.schedule(reschedule_delay, self.deploy) - else: - # TODO: Create socket!! - socket_name = self.get("socketName") - self._client = LinuxNS3Client(socket_name) - - #self.do_discover() - #self.do_provision() - - self.set_ready() - - def do_start(self): - command = self.get("command") - - self.info("Starting ns-3 simulation") - - self._client.start() - - self.set_started() - - # TODO! DESTROY SOCKET AND SHUTDOWN CLIENT diff --git a/src/nepi/resources/ns3/ns3application.py b/src/nepi/resources/ns3/ns3application.py index b00b47f9..c30a00c5 100644 --- a/src/nepi/resources/ns3/ns3application.py +++ b/src/nepi/resources/ns3/ns3application.py @@ -45,29 +45,40 @@ class NS3BaseApplication(NS3Base): def _connect_object(self): node = self.node if node.uuid not in self.connected: - self.simulator.invoke(node.uuid, "AddApplication", self.uuid) + self.simulation.invoke(node.uuid, "AddApplication", self.uuid) self._connected.add(node.uuid) - """ - def do_start(self): - if self.state == ResourceState.READY: - self.info("Starting") - - # BUG: without doing this explicit call it doesn't start!!! - # Shouldn't be enough to set the StartTime? - self.simulator.invoke(self.uuid, "Start") - - self.set_started() - else: - msg = " Failed " - self.error(msg, out, err) - raise RuntimeError, msg def do_stop(self): if self.state == ResourceState.STARTED: - # No need to do anything, simulator.Destroy() will stop every object + # No need to do anything, simulation.Destroy() will stop every object self.info("Stopping command '%s'" % command) - self.simulator.invoke(self.uuid, "Stop") + self.simulation.invoke(self.uuid, "Stop") self.set_stopped() - """ + + @property + def state(self): + #if self._state == ResourceState.STARTED: + + return self._state + """ + now = testbed_instance.ns3.Simulator.Now() + if now.IsZero(): + return STATUS_NOT_STARTED + app = testbed_instance.elements[guid] + parameters = testbed_instance._get_parameters(guid) + if "StartTime" in parameters and parameters["StartTime"]: + start_value = parameters["StartTime"] + start_time = testbed_instance.ns3.Time(start_value) + if now.Compare(start_time) < 0: + return STATUS_NOT_STARTED + if "StopTime" in parameters and parameters["StopTime"]: + stop_value = parameters["StopTime"] + stop_time = testbed_instance.ns3.Time(stop_value) + if now.Compare(stop_time) < 0: + return STATUS_RUNNING + else: + return STATUS_FINISHED + return STATUS_UNDETERMINED + """ diff --git a/src/nepi/resources/ns3/ns3base.py b/src/nepi/resources/ns3/ns3base.py index 7cc077a5..adcbb8b4 100644 --- a/src/nepi/resources/ns3/ns3base.py +++ b/src/nepi/resources/ns3/ns3base.py @@ -22,6 +22,8 @@ from nepi.execution.resource import ResourceManager, clsinit_copy, \ from nepi.execution.attribute import Flags +reschedule_delay = "2s" + @clsinit_copy class NS3Base(ResourceManager): _rtype = "abstract::ns3::Object" @@ -41,8 +43,8 @@ class NS3Base(ResourceManager): return self._uuid @property - def simulator(self): - return self.node.simulator + def simulation(self): + return self.node.simulation @property def node(self): @@ -76,7 +78,7 @@ class NS3Base(ResourceManager): kwargs[attr.name] = attr.value - self._uuid = self.simulator.factory(self.get_rtype(), **kwargs) + self._uuid = self.simulation.factory(self.get_rtype(), **kwargs) def _configure_object(self): pass @@ -84,20 +86,19 @@ class NS3Base(ResourceManager): def _connect_object(self): node = self.node if node and node.uuid not in self.connected: - self.simulator.invoke(node.uuid, "AggregateObject", self.uuid) + self.simulation.invoke(node.uuid, "AggregateObject", self.uuid) self._connected.add(node.uuid) def _wait_rms(self): """ Returns True if dependent RMs are not yer READY, False otherwise""" for rm in self._rms_to_wait: if rm and rm.state < ResourceState.READY: - rm.debug("Not yet READY") return True return False def do_provision(self): # TODO: create run dir for ns3 object !!!! - # self.simulator.node.mkdir(self.run_home) + # self.simulation.node.mkdir(self.run_home) self._instantiate_object() self._connect_object() @@ -120,7 +121,7 @@ class NS3Base(ResourceManager): def do_start(self): if self.state == ResourceState.READY: - # No need to do anything, simulator.Run() will start every object + # No need to do anything, simulation.Run() will start every object self.info("Starting") self.set_started() else: @@ -130,7 +131,7 @@ class NS3Base(ResourceManager): def do_stop(self): if self.state == ResourceState.STARTED: - # No need to do anything, simulator.Destroy() will stop every object + # No need to do anything, simulation.Destroy() will stop every object self.info("Stopping command '%s'" % command) self.set_stopped() diff --git a/src/nepi/resources/ns3/ns3channel.py b/src/nepi/resources/ns3/ns3channel.py index 92dc8928..85556fe7 100644 --- a/src/nepi/resources/ns3/ns3channel.py +++ b/src/nepi/resources/ns3/ns3channel.py @@ -25,8 +25,8 @@ class NS3BaseChannel(NS3Base): _rtype = "abstract::ns3::Channel" @property - def simulator(self): - return self.devices[0].node.simulator + def simulation(self): + return self.devices[0].node.simulation @property def devices(self): @@ -43,6 +43,6 @@ class NS3BaseChannel(NS3Base): @property def _rms_to_wait(self): rms = set() - rms.add(self.simulator) + rms.add(self.simulation) return rms diff --git a/src/nepi/resources/ns3/ns3errormodel.py b/src/nepi/resources/ns3/ns3errormodel.py index fa0779da..d46293f3 100644 --- a/src/nepi/resources/ns3/ns3errormodel.py +++ b/src/nepi/resources/ns3/ns3errormodel.py @@ -45,6 +45,6 @@ class NS3BaseErrorModel(NS3Base): def _connect_object(self): device = self.device if device.uuid not in self.connected: - self.simulator.invoke(device.uuid, "SetReceiveErrorModel", self.uuid) + self.simulation.invoke(device.uuid, "SetReceiveErrorModel", self.uuid) self._connected.add(device.uuid) diff --git a/src/nepi/resources/ns3/ns3errorratemodel.py b/src/nepi/resources/ns3/ns3errorratemodel.py index f1b283a4..f1f56668 100644 --- a/src/nepi/resources/ns3/ns3errorratemodel.py +++ b/src/nepi/resources/ns3/ns3errorratemodel.py @@ -45,6 +45,6 @@ class NS3BaseErrorRateModel(NS3Base): def _connect_object(self): phy = self.phy if phy.uuid not in self.connected: - self.simulator.invoke(phy.uuid, "SetErrorRateModel", self.uuid) + self.simulation.invoke(phy.uuid, "SetErrorRateModel", self.uuid) self._connected.add(phy.uuid) diff --git a/src/nepi/resources/ns3/ns3ipv4l3protocol.py b/src/nepi/resources/ns3/ns3ipv4l3protocol.py index f8f3670b..526cae7d 100644 --- a/src/nepi/resources/ns3/ns3ipv4l3protocol.py +++ b/src/nepi/resources/ns3/ns3ipv4l3protocol.py @@ -43,11 +43,11 @@ class NS3BaseIpv4L3Protocol(NS3Base): return rms def _configure_object(self): - simulator = self.simulator + simulation = self.simulation - uuid_list_routing = simulator.create("Ipv4ListRouting") - simulator.invoke(self.uuid, "SetRoutingProtocol", uuid_list_routing) + uuid_list_routing = simulation.create("Ipv4ListRouting") + simulation.invoke(self.uuid, "SetRoutingProtocol", uuid_list_routing) - uuid_static_routing = simulator.create("Ipv4StaticRouting") - simulator.invoke(uuid_list_routing, "AddRoutingProtocol", uuid_static_routing, 1) + uuid_static_routing = simulation.create("Ipv4StaticRouting") + simulation.invoke(uuid_list_routing, "AddRoutingProtocol", uuid_static_routing, 1) diff --git a/src/nepi/resources/ns3/ns3netdevice.py b/src/nepi/resources/ns3/ns3netdevice.py index 66366cbd..87c01df1 100644 --- a/src/nepi/resources/ns3/ns3netdevice.py +++ b/src/nepi/resources/ns3/ns3netdevice.py @@ -84,10 +84,10 @@ class NS3BaseNetDevice(NS3Base): # Set Mac mac = self.get("mac") if mac: - mac_uuid = self.simulator.create("Mac48Address", mac) + mac_uuid = self.simulation.create("Mac48Address", mac) else: - mac_uuid = self.simulator.invoke("singleton::Mac48Address", "Allocate") - self.simulator.invoke(self.uuid, "SetAddress", mac_uuid) + mac_uuid = self.simulation.invoke("singleton::Mac48Address", "Allocate") + self.simulation.invoke(self.uuid, "SetAddress", mac_uuid) # Set IP address ip = self.get("ip") @@ -97,16 +97,16 @@ class NS3BaseNetDevice(NS3Base): if i.version == 4: # IPv4 ipv4 = self.node.ipv4 - ifindex_uuid = self.simulator.invoke(ipv4.uuid, "AddInterface", + ifindex_uuid = self.simulation.invoke(ipv4.uuid, "AddInterface", self.uuid) - ipv4_addr_uuid = self.simulator.create("Ipv4Address", ip) - ipv4_mask_uuid = self.simulator.create("Ipv4Mask", "/%s" % str(prefix)) - inaddr_uuid = self.simulator.create("Ipv4InterfaceAddress", + ipv4_addr_uuid = self.simulation.create("Ipv4Address", ip) + ipv4_mask_uuid = self.simulation.create("Ipv4Mask", "/%s" % str(prefix)) + inaddr_uuid = self.simulation.create("Ipv4InterfaceAddress", ipv4_addr_uuid, ipv4_mask_uuid) - self.simulator.invoke(ipv4.uuid, "AddAddress", ifindex_uuid, + self.simulation.invoke(ipv4.uuid, "AddAddress", ifindex_uuid, inaddr_uuid) - self.simulator.invoke(ipv4.uuid, "SetMetric", ifindex_uuid, 1) - self.simulator.invoke(ipv4.uuid, "SetUp", ifindex_uuid) + self.simulation.invoke(ipv4.uuid, "SetMetric", ifindex_uuid, 1) + self.simulation.invoke(ipv4.uuid, "SetUp", ifindex_uuid) else: # IPv6 # TODO! @@ -115,11 +115,11 @@ class NS3BaseNetDevice(NS3Base): def _connect_object(self): node = self.node if node and node.uuid not in self.connected: - self.simulator.invoke(node.uuid, "AddDevice", self.uuid) + self.simulation.invoke(node.uuid, "AddDevice", self.uuid) self._connected.add(node.uuid) channel = self.channel if channel and channel.uuid not in self.connected: - self.simulator.invoke(self.uuid, "Attach", channel.uuid) + self.simulation.invoke(self.uuid, "Attach", channel.uuid) self._connected.add(channel.uuid) diff --git a/src/nepi/resources/ns3/ns3node.py b/src/nepi/resources/ns3/ns3node.py index 62d2a38d..3f1ce33f 100644 --- a/src/nepi/resources/ns3/ns3node.py +++ b/src/nepi/resources/ns3/ns3node.py @@ -19,20 +19,20 @@ from nepi.execution.resource import clsinit_copy from nepi.resources.ns3.ns3base import NS3Base -from nepi.resources.ns3.ns3simulator import NS3Simulator @clsinit_copy class NS3BaseNode(NS3Base): _rtype = "abstract::ns3::Node" @property - def simulator(self): + def simulation(self): + from nepi.resources.ns3.ns3simulation import NS3Simulation for guid in self.connections: rm = self.ec.get_resource(guid) - if isinstance(rm, NS3Simulator): + if isinstance(rm, NS3Simulation): return rm - msg = "Node not connected to simulator" + msg = "Node not connected to simulation" self.error(msg) raise RuntimeError, msg @@ -46,11 +46,11 @@ class NS3BaseNode(NS3Base): @property def _rms_to_wait(self): rms = set() - rms.add(self.simulator) + rms.add(self.simulation) return rms def _configure_object(self): ### node.AggregateObject(PacketSocketFactory()) - uuid_packet_socket_factory = self.simulator.create("PacketSocketFactory") - self.simulator.invoke(self.uuid, "AggregateObject", uuid_packet_socket_factory) + uuid_packet_socket_factory = self.simulation.create("PacketSocketFactory") + self.simulation.invoke(self.uuid, "AggregateObject", uuid_packet_socket_factory) diff --git a/src/nepi/resources/ns3/ns3propagationdelaymodel.py b/src/nepi/resources/ns3/ns3propagationdelaymodel.py index 6c5ee861..1a5b72a2 100644 --- a/src/nepi/resources/ns3/ns3propagationdelaymodel.py +++ b/src/nepi/resources/ns3/ns3propagationdelaymodel.py @@ -25,8 +25,8 @@ class NS3BasePropagationDelayModel(NS3Base): _rtype = "abstract::ns3::PropagationDelayModel" @property - def simulator(self): - return self.channel.simulator + def simulation(self): + return self.channel.simulation @property def channel(self): diff --git a/src/nepi/resources/ns3/ns3propagationlossmodel.py b/src/nepi/resources/ns3/ns3propagationlossmodel.py index fdaef388..8b4a0856 100644 --- a/src/nepi/resources/ns3/ns3propagationlossmodel.py +++ b/src/nepi/resources/ns3/ns3propagationlossmodel.py @@ -25,8 +25,8 @@ class NS3BasePropagationLossModel(NS3Base): _rtype = "ns3::PropagationLossModel" @property - def simulator(self): - return self.channel.simulator + def simulation(self): + return self.channel.simulation @property def channel(self): @@ -49,6 +49,6 @@ class NS3BasePropagationLossModel(NS3Base): def _connect_object(self): channel = self.channel if channel.uuid not in self.connected: - self.simulator.invoke(channel.uuid, "SetPropagationLossModel", self.uuid) + self.simulation.invoke(channel.uuid, "SetPropagationLossModel", self.uuid) self._connected.add(channel.uuid) diff --git a/src/nepi/resources/ns3/ns3queue.py b/src/nepi/resources/ns3/ns3queue.py index 9fd8df9f..d7ec2b7c 100644 --- a/src/nepi/resources/ns3/ns3queue.py +++ b/src/nepi/resources/ns3/ns3queue.py @@ -49,6 +49,6 @@ class NS3BaseQueue(NS3Base): def _connect_object(self): device = self.device if device.uuid not in self.connected: - self.simulator.invoke(device.uuid, "SetQueue", self.uuid) + self.simulation.invoke(device.uuid, "SetQueue", self.uuid) self._connected.add(device.uuid) diff --git a/src/nepi/resources/ns3/ns3server.py b/src/nepi/resources/ns3/ns3server.py index 9e3920f4..becf6057 100644 --- a/src/nepi/resources/ns3/ns3server.py +++ b/src/nepi/resources/ns3/ns3server.py @@ -23,6 +23,7 @@ import errno import logging import os import socket +import sys from optparse import OptionParser, SUPPRESS_HELP @@ -156,7 +157,7 @@ def send_reply(conn, reply): conn.send("%s\n" % encoded) def get_options(): - usage = ("usage: %prog -S -L -v ") + usage = ("usage: %prog -S -L -H -v ") parser = OptionParser(usage = usage) @@ -168,6 +169,10 @@ def get_options(): help = "NS_LOG environmental variable to be set", default = "", type="str") + parser.add_option("-H", "--homedir", dest="homedir", + help = "Home directory where to store results", + default = "", type="str") + parser.add_option("-v", "--verbose", help="Print debug output", action="store_true", @@ -175,11 +180,11 @@ def get_options(): (options, args) = parser.parse_args() - return options.socket_name, options.verbose, options.ns_log - -def run_server(socket_name, verbose = False, ns_log = None): + return (options.socket_name, options.homedir, options.verbose, + options.ns_log) - level = logging.DEBUG if verbose else logging.INFO +def run_server(socket_name, homedir = None, level = logging.INFO, + ns_log = None): # Sets NS_LOG environmental variable for NS debugging if ns_log: @@ -187,7 +192,9 @@ def run_server(socket_name, verbose = False, ns_log = None): ###### ns-3 wrapper instantiation - ns3_wrapper = NS3Wrapper(loglevel=level) + ns3_wrapper = NS3Wrapper(homedir = homedir, loglevel=level) + + ns3_wrapper.logger.info("STARTING...") # create unix socket to receive instructions sock = create_socket(socket_name) @@ -212,17 +219,38 @@ def run_server(socket_name, verbose = False, ns_log = None): if msg == NS3WrapperMessage.SHUTDOWN: stop = True - - reply = handle_message(ns3_wrapper, msg, args) + + try: + reply = handle_message(ns3_wrapper, msg, args) + except: + import traceback + err = traceback.format_exc() + ns3_wrapper.logger.error(err) + raise try: send_reply(conn, reply) except socket.error: break + + ns3_wrapper.logger.info("EXITING...") if __name__ == '__main__': - (socket_name, verbose, ns_log) = get_options() + (socket_name, homedir, verbose, ns_log) = get_options() + + ## configure logging + FORMAT = "%(asctime)s %(name)s %(levelname)-4s %(message)s" + level = logging.DEBUG if verbose else logging.INFO + + logging.basicConfig(format = FORMAT, level = level) + + # Make sure to send DEBUG messages to stdout instead of stderr + root = logging.getLogger() + handler = logging.StreamHandler(sys.stdout) + handler.setLevel(logging.DEBUG) + root.addHandler(handler) - run_server(socket_name, verbose, ns_log) + ## Run the server + run_server(socket_name, homedir, level, ns_log) diff --git a/src/nepi/resources/ns3/ns3simulator.py b/src/nepi/resources/ns3/ns3simulation.py similarity index 98% rename from src/nepi/resources/ns3/ns3simulator.py rename to src/nepi/resources/ns3/ns3simulation.py index d0f14189..2baf1a34 100644 --- a/src/nepi/resources/ns3/ns3simulator.py +++ b/src/nepi/resources/ns3/ns3simulation.py @@ -17,7 +17,7 @@ # # Author: Alina Quereilhac -class NS3Simulator(object): +class NS3Simulation(object): @property def client(self): return self._client diff --git a/src/nepi/resources/ns3/ns3wifimac.py b/src/nepi/resources/ns3/ns3wifimac.py index 7e1cf8f8..700f6cb4 100644 --- a/src/nepi/resources/ns3/ns3wifimac.py +++ b/src/nepi/resources/ns3/ns3wifimac.py @@ -49,6 +49,6 @@ class NS3BaseWifiMac(NS3Base): def _connect_object(self): device = self.device if device.uuid not in self.connected: - self.simulator.invoke(device.uuid, "SetMac", self.uuid) + self.simulation.invoke(device.uuid, "SetMac", self.uuid) self._connected.add(device.uuid) diff --git a/src/nepi/resources/ns3/ns3wifiphy.py b/src/nepi/resources/ns3/ns3wifiphy.py index 936eddc8..8b6166ac 100644 --- a/src/nepi/resources/ns3/ns3wifiphy.py +++ b/src/nepi/resources/ns3/ns3wifiphy.py @@ -49,7 +49,7 @@ class NS3BaseWifiPhy(NS3Base): def _connect_object(self): device = self.device if device.uuid not in self.connected: - self.simulator.invoke(device.uuid, "SetPhy", self.uuid) + self.simulation.invoke(device.uuid, "SetPhy", self.uuid) self.simulator.invoke(self.uuid, "SetDevice", device.uuid) self._connected.add(device.uuid) diff --git a/src/nepi/resources/ns3/ns3wifiremotestationmanager.py b/src/nepi/resources/ns3/ns3wifiremotestationmanager.py index a5f4548e..49eb52f0 100644 --- a/src/nepi/resources/ns3/ns3wifiremotestationmanager.py +++ b/src/nepi/resources/ns3/ns3wifiremotestationmanager.py @@ -50,6 +50,6 @@ class NS3BaseWifiRemoteStationManager(NS3Base): def _connect_object(self): device = self.device if device.uuid not in self.connected: - self.simulator.invoke(device.uuid, "SetRemoteStationManager", self.uuid) + self.simulation.invoke(device.uuid, "SetRemoteStationManager", self.uuid) self._connected.add(device.uuid) diff --git a/src/nepi/resources/ns3/ns3wrapper.py b/src/nepi/resources/ns3/ns3wrapper.py index 47525f96..9f84db70 100644 --- a/src/nepi/resources/ns3/ns3wrapper.py +++ b/src/nepi/resources/ns3/ns3wrapper.py @@ -39,7 +39,9 @@ def load_ns3_module(): regex = re.compile("(.*\.so)$") libs = [m.group(1) for filename in files for m in [regex.search(filename)] if m] - libscp = list(libs) + initial_size = len(libs) + # Try to load the libraries in the right order by trial and error. + # Loop until all libraries are loaded. while len(libs) > 0: for lib in libs: libfile = os.path.join(libdir, lib) @@ -47,13 +49,16 @@ def load_ns3_module(): ctypes.CDLL(libfile, ctypes.RTLD_GLOBAL) libs.remove(lib) except: + #import traceback + #err = traceback.format_exc() + #print err pass # if did not load any libraries in the last iteration break # to prevent infinit loop - if len(libscp) == len(libs): + if initial_size == len(libs): raise RuntimeError("Imposible to load shared libraries %s" % str(libs)) - libscp = list(libs) + initial_size = list(libs) # import the python bindings for the ns-3 modules if bindings: @@ -63,11 +68,14 @@ def load_ns3_module(): import imp import ns - # create a module to add all ns3 classes + # create a Python module to add all ns3 classes ns3mod = imp.new_module("ns3") sys.modules["ns3"] = ns3mod for importer, modname, ispkg in pkgutil.iter_modules(ns.__path__): + if modname in [ "visualizer" ]: + continue + fullmodname = "ns.%s" % modname module = __import__(fullmodname, globals(), locals(), ['*']) @@ -109,12 +117,6 @@ class NS3Wrapper(object): # Logging self._logger = logging.getLogger("ns3wrapper") self._logger.setLevel(loglevel) - - hdlr = logging.FileHandler(os.path.join(self.homedir, "ns3wrapper.log")) - formatter = logging.Formatter('%(asctime)s %(levelname)s %(message)s') - hdlr.setFormatter(formatter) - - self._logger.addHandler(hdlr) ## NOTE that the reason to create a handler to the ns3 module, # that is re-loaded each time a ns-3 wrapper is instantiated, @@ -146,7 +148,6 @@ class NS3Wrapper(object): tid_count = type_id.GetRegisteredN() base = type_id.LookupByName("ns3::Object") - # Create a .py file using the ns-3 RM template for each ns-3 TypeId for i in xrange(tid_count): tid = type_id.GetRegistered(i) diff --git a/src/nepi/util/execfuncs.py b/src/nepi/util/execfuncs.py index cd11bfae..98b9d739 100644 --- a/src/nepi/util/execfuncs.py +++ b/src/nepi/util/execfuncs.py @@ -24,7 +24,6 @@ import subprocess def lexec(command, user = None, sudo = False, - stdin = None, env = None): """ Executes a local command, returns ((stdout,stderr),process) @@ -44,7 +43,6 @@ def lexec(command, proc = subprocess.Popen(command, shell=True, stdout = subprocess.PIPE, stderr = subprocess.PIPE) - #stdin = stdin) out, err = proc.communicate() return ((out, err), proc) diff --git a/src/nepi/util/sshfuncs.py b/src/nepi/util/sshfuncs.py index 8c8ccf11..458af495 100644 --- a/src/nepi/util/sshfuncs.py +++ b/src/nepi/util/sshfuncs.py @@ -206,11 +206,82 @@ def eintr_retry(func): return func(*p, **kw) return rv +def socat(local_socket_name, remote_socket_name, + host, user, + port = None, + agent = True, + sudo = False, + identity = None, + server_key = None, + env = None, + tty = False, + connect_timeout = 30, + retry = 3, + strict_host_checking = True): + """ + Executes a remote command, returns ((stdout,stderr),process) + """ + + tmp_known_hosts = None + hostip = gethostbyname(host) + + + args = ["socat"] + args.append("UNIX-LISTEN:%s,unlink-early,fork" % local_socket_name) + + ssh_args = ['ssh', '-C', + # Don't bother with localhost. Makes test easier + '-o', 'NoHostAuthenticationForLocalhost=yes', + '-o', 'ConnectTimeout=%d' % (int(connect_timeout),), + '-o', 'ConnectionAttempts=3', + '-o', 'ServerAliveInterval=30', + '-o', 'TCPKeepAlive=yes', + '-l', user, hostip or host] + + if not strict_host_checking: + # Do not check for Host key. Unsafe. + ssh_args.extend(['-o', 'StrictHostKeyChecking=no']) + + if agent: + ssh_args.append('-A') + + if port: + ssh_args.append('-p%d' % port) + + if identity: + ssh_args.extend(('-i', identity)) + + if tty: + ssh_args.append('-t') + ssh_args.append('-t') + + if server_key: + # Create a temporary server key file + tmp_known_hosts = make_server_key_args(server_key, host, port) + ssh_args.extend(['-o', 'UserKnownHostsFile=%s' % (tmp_known_hosts.name,)]) + + ssh_cmd = " ".join(ssh_args) + + exec_cmd = "EXEC:'%s socat STDIO UNIX-CONNECT\:%s'" % (ssh_cmd, + remote_socket_name) + + args.append(exec_cmd) + + log_msg = " socat - host %s - command %s " % (host, " ".join(args)) + + return _retry_rexec(args, log_msg, + stdout = None, + stdin = None, + stderr = None, + env = env, + retry = retry, + tmp_known_hosts = tmp_known_hosts, + blocking = False) + def rexec(command, host, user, port = None, agent = True, sudo = False, - stdin = None, identity = None, server_key = None, env = None, @@ -275,7 +346,16 @@ def rexec(command, host, user, log_msg = " rexec - host %s - command %s " % (host, " ".join(args)) - return _retry_rexec(args, log_msg, env = env, retry = retry, + stdout = stderr = stdin = subprocess.PIPE + if forward_x11: + stdout = stderr = stdin = None + + return _retry_rexec(args, log_msg, + stderr = stderr, + stdin = stdin, + stdout = stdout, + env = env, + retry = retry, tmp_known_hosts = tmp_known_hosts, blocking = blocking) @@ -338,7 +418,7 @@ def rcopy(source, dest, # Do not check for Host key. Unsafe. args.extend(['-o', 'StrictHostKeyChecking=no']) - if isinstance(source,list): + if isinstance(source, list): args.extend(source) else: if openssh_has_persist(): @@ -614,6 +694,9 @@ fi def _retry_rexec(args, log_msg, + stdout = subprocess.PIPE, + stdin = subprocess.PIPE, + stderr = subprocess.PIPE, env = None, retry = 3, tmp_known_hosts = None, @@ -623,9 +706,9 @@ def _retry_rexec(args, # connects to the remote host and starts a remote connection proc = subprocess.Popen(args, env = env, - stdout = subprocess.PIPE, - stdin = subprocess.PIPE, - stderr = subprocess.PIPE) + stdout = stdout, + stdin = stdin, + stderr = stderr) # attach tempfile object to the process, to make sure the file stays # alive until the process is finished with it @@ -634,11 +717,13 @@ def _retry_rexec(args, # The argument block == False forces to rexec to return immediately, # without blocking try: + err = out = " " if blocking: (out, err) = proc.communicate() - else: - err = proc.stderr.read() + elif stdout: out = proc.stdout.read() + if proc.poll() and stderr: + err = proc.stderr.read() log(log_msg, logging.DEBUG, out, err) diff --git a/test/resources/linux/ns3/ns3simulator.py b/test/resources/linux/ns3/ns3simulation.py similarity index 81% rename from test/resources/linux/ns3/ns3simulator.py rename to test/resources/linux/ns3/ns3simulation.py index a4a97e0b..cf08f0e3 100644 --- a/test/resources/linux/ns3/ns3simulator.py +++ b/test/resources/linux/ns3/ns3simulation.py @@ -30,40 +30,28 @@ from nepi.execution.ec import ExperimentController -from nepi.resources.ns3.ns3server import run_server import os -import threading import time import unittest class LinuxNS3ClientTest(unittest.TestCase): - def setUp(self): - self.socket_name = os.path.join("/", "tmp", "NS3WrapperServerSimu.sock") - if os.path.exists(self.socket_name): - os.remove(self.socket_name) - - def tearDown(self): - os.remove(self.socket_name) - def test_runtime_attr_modify(self): - thread = threading.Thread(target = run_server, - args = [self.socket_name], - kwargs = {"verbose" : True, - "ns_log": "V4Ping:Node"}) - - thread.setDaemon(True) - thread.start() - - time.sleep(1) + ssh_key = "%s/.ssh/id_rsa_planetlab" % (os.environ['HOME']) ec = ExperimentController(exp_id = "test-ns3-simu") node = ec.register_resource("LinuxNode") - ec.set(node, "hostname", "localhost") - - simu = ec.register_resource("LinuxNS3Simulator") - ec.set(simu, "socketName", self.socket_name) + #ec.set(node, "hostname", "roseval.pl.sophia.inria.fr") + #ec.set(node, "username", "alina") + ec.set(node, "hostname", "peeramide.irisa.fr") + #ec.set(node, "hostname", "planetlab2.upc.es") + ec.set(node, "username", "inria_alina") + ec.set(node, "identity", ssh_key) + ec.set(node, "cleanProcesses", True) + + simu = ec.register_resource("LinuxNS3Simulation") + ec.set(simu, "verbose", True) ec.register_connection(simu, node) nsnode1 = ec.register_resource("ns3::Node") @@ -121,10 +109,11 @@ class LinuxNS3ClientTest(unittest.TestCase): ec.deploy() - time.sleep(30) + #time.sleep(60) + ec.wait_started([ping]) + #ec.wait_finised([ping]) ec.shutdown() - if __name__ == '__main__': unittest.main()