Bugfixing for ns3::MatrixPropagationLossModel ...
[nepi.git] / src / nepi / testbeds / ns3 / execute.py
index 6bb58dc..7de7bde 100644 (file)
@@ -1,18 +1,74 @@
-#!/usr/bin/env python
 # -*- coding: utf-8 -*-
 
-from constants import TESTBED_ID
+from util import  _get_ipv4_protocol_guid, _get_node_guid, _get_dev_number
 from nepi.core import testbed_impl
 from nepi.core.attributes import Attribute
-from nepi.util.constants import AF_INET, AF_INET6
+from constants import TESTBED_ID, TESTBED_VERSION
+from nepi.util.constants import TIME_NOW, TestbedStatus as TS
 import os
+import sys
+import threading
+import random
+import socket
+import weakref
 
-class TestbedInstance(testbed_impl.TestbedInstance):
-    def __init__(self, testbed_version):
-        super(TestbedInstance, self).__init__(TESTBED_ID, testbed_version)
+def load_ns3_module():
+    import sys
+    if 'ns3' in sys.modules:
+        return
+
+    import ctypes
+    import imp
+    import re
+
+    bindings = os.environ["NEPI_NS3BINDINGS"] \
+               if "NEPI_NS3BINDINGS" in os.environ else None
+    libdir = os.environ["NEPI_NS3LIBRARY"] \
+               if "NEPI_NS3LIBRARY" in os.environ else None
+
+    if libdir:
+        files = os.listdir(libdir)
+        regex = re.compile("(.*\.so)$")
+        libs = [m.group(1) for filename in files for m in [regex.search(filename)] if m]
+
+        libscp = list(libs)
+        while len(libs) > 0:
+            for lib in libscp:
+                libfile = os.path.join(libdir, lib)
+                try:
+                    ctypes.CDLL(libfile, ctypes.RTLD_GLOBAL)
+                    libs.remove(lib)
+                except:
+                    pass
+            # if did not load any libraries in the last iteration
+            if len(libscp) == len(libs):
+                raise RuntimeError("Imposible to load shared libraries %s" % str(libs))
+            libscp = list(libs)
+
+    if bindings:
+        sys.path.append(bindings)
+
+    import ns3_bindings_import as mod
+    sys.modules["ns3"] = mod
+
+class TestbedController(testbed_impl.TestbedController):
+    from nepi.util.tunchannel_impl import TunChannel
+    
+    LOCAL_FACTORIES = {
+        'ns3::Nepi::TunChannel' : TunChannel,
+    }
+   
+    DUMMY_FACTORIES = ['ns3::Nepi::MobilityPair']
+
+    LOCAL_TYPES = tuple(LOCAL_FACTORIES.values())
+
+    def __init__(self):
+        super(TestbedController, self).__init__(TESTBED_ID, TESTBED_VERSION)
         self._ns3 = None
         self._home_directory = None
         self._traces = dict()
+        self._simulator_thread = None
+        self._condition = None
 
     @property
     def home_directory(self):
@@ -25,96 +81,100 @@ class TestbedInstance(testbed_impl.TestbedInstance):
     def do_setup(self):
         self._home_directory = self._attributes.\
             get_attribute_value("homeDirectory")
-        self._ns3 = self._load_ns3_module()
-
-    def do_configure(self):
-        # configure addressess
-        for guid, addresses in self._add_address.iteritems():
-            element = self._elements[guid]
-            for address in addresses:
-                (family, address, netprefix, broadcast) = address
-                if family == AF_INET:
-                    pass
-                    # TODO!!!
-        # configure routes
-        for guid, routes in self._add_route.iteritems():
-            element = self._elements[guid]
-            for route in routes:
-                (destination, netprefix, nexthop) = route
-                # TODO!!
-        """
-        context = self.server.modules.ns3
-        ipv4 = self._object
-        for interface in self._interface2addr:
-            ifindex = ipv4.AddInterface(interface._object)
-            for addr in self._interface2addr[interface]:
-                inaddr = context.Ipv4InterfaceAddress(
-                        context.Ipv4Address(
-                            addr.get_attribute("Address").value),
-                        context.Ipv4Mask(
-                            addr.get_attribute("NetPrefix").value))
-                ipv4.AddAddress(ifindex, inaddr)
-                ipv4.SetMetric(ifindex, 1)
-                ipv4.SetUp(ifindex)
-                self._interface_addrs[addr] = inaddr
-                self._interfaces[interface] = ifindex
-        for entry in self.get_node().routing_table.get_entries(self._af):
-            self._rt_add(entry)
-
-        def _rt_add(self, entry):
-        # Called both at install-time (by NS3Ipv4Stack.post_install) and at
-        # run-time (by RoutingTable.add_entry).
-        context = self.server.modules.ns3
-        ifindex = self._interfaces[entry.interface]
-        prefixlen = entry.prefixlen
-        # print "rt_add %s %s %s %d"% (prefix, prefixlen, entry.nexthop, ifindex)        
-        if entry.nexthop:
-            self._static_routing.AddNetworkRouteTo(
-                    context.Ipv4Address(entry.prefix.address),
-                    context.Ipv4Mask(entry.mask.address),
-                    context.Ipv4Address(entry.nexthop.address),
-                    ifindex)
-        else:
-            self._static_routing.AddNetworkRouteTo(
-                    context.Ipv4Address(entry.prefix.address),
-                    context.Ipv4Mask(entry.mask.address),
-                    ifindex)
-        """
-
-    def set(self, time, guid, name, value):
-        super(TestbedInstance, self).set(time, guid, name, value)
+        self._ns3 = self._configure_ns3_module()
+        
+        # create home...
+        home = os.path.normpath(self.home_directory)
+        if not os.path.exists(home):
+            os.makedirs(home, 0755)
+        
+        super(TestbedController, self).do_setup()
+
+    def start(self):
+        super(TestbedController, self).start()
+        self._condition = threading.Condition()
+        self._simulator_thread = threading.Thread(target = self._simulator_run,
+                args = [self._condition])
+        self._simulator_thread.setDaemon(True)
+        self._simulator_thread.start()
+
+    def stop(self, time = TIME_NOW):
+        super(TestbedController, self).stop(time)
+        self._stop_simulation(time)
+
+    def set(self, guid, name, value, time = TIME_NOW):
+        super(TestbedController, self).set(guid, name, value, time)
+
         # TODO: take on account schedule time for the task
-        factory_id = self._crerate[guid]
+        factory_id = self._create[guid]
+        if factory_id in self.DUMMY_FACTORIES:
+            return 
+
+        factory = self._factories[factory_id]
         element = self._elements[guid]
-        TypeId = self.ns3.TypeId()
-        typeid = TypeId.LookupByName(factory_id)
-        info = TypeId.AttributeInfo()
-        if not typeid.LookupAttributeByName(name, info):
-            raise RuntimeError("Attribute %s doesn't belong to element %s" \
-                   % (name, factory_id))
-        value = str(value)
-        if isinstance(value, bool):
-            value = value.lower()
-        ns3_value = info.checker.Create()
-        ns3_value.DeserializeFromString(value, checker)
-        element.SetAttribute(name, ns3_value)
+        if factory_id in self.LOCAL_FACTORIES:
+            setattr(element, name, value)
+        elif not factory.box_attributes.is_attribute_metadata(name):
+            if name == "Up":
+                ipv4_guid =  _get_ipv4_protocol_guid(self, guid)
+                if not ipv4_guid in self._elements:
+                    return
+                ipv4 = self._elements[ipv4_guid]
+                if value == False:
+                    nint = ipv4.GetNInterfaces()
+                    for i in xrange(0, nint):
+                        ipv4.SetDown(i)
+                else:
+                    nint = ipv4.GetNInterfaces()
+                    for i in xrange(0, nint):
+                        ipv4.SetUp(i)
+            else:
+                ns3_value = self._to_ns3_value(guid, name, value)
+                self._set_attribute(name, ns3_value, element)
+
+    def get(self, guid, name, time = TIME_NOW):
+        value = super(TestbedController, self).get(guid, name, time)
 
-    def get(self, time, guid, name):
         # TODO: take on account schedule time for the task
+        factory_id = self._create[guid]
+        if factory_id in self.DUMMY_FACTORIES:
+            return 
+
+        factory = self._factories[factory_id]
+        element = self._elements[guid]
+        if factory_id in self.LOCAL_FACTORIES:
+            if hasattr(element, name):
+                return getattr(element, name)
+            else:
+                return value
+        else: 
+            if name == "Up":
+                ipv4_guid =  _get_ipv4_protocol_guid(self, guid)
+                if not ipv4_guid in self._elements:
+                    return True
+                ipv4 = self._elements[ipv4_guid]
+                nint = ipv4.GetNInterfaces()
+                value = True
+                for i in xrange(0, nint):
+                    value = ipv4.IsUp(i)
+                    if not value: break
+                return value
+
+        if factory.box_attributes.is_attribute_metadata(name):
+            return value
+
         TypeId = self.ns3.TypeId()
         typeid = TypeId.LookupByName(factory_id)
-        info = TypeId.AttributeInfo()
-        if not typeid.LookupAttributeByName(name, info):
-            raise RuntimeError("Attribute %s doesn't belong to element %s" \
-                   % (name, factory_id))
+        info = TypeId.AttributeInformation()
+        if not typeid or not typeid.LookupAttributeByName(name, info):
+            raise AttributeError("Invalid attribute %s for element type %d" % \
+                (name, guid))
         checker = info.checker
         ns3_value = checker.Create() 
-        element = self._elements[guid]
-        element.GetAttribute(name, ns3_value)
+        self._get_attribute(name, ns3_value, element)
         value = ns3_value.SerializeToString(checker)
-        factory_id = self._create[guid]
-        factory = self._factories[factory_id]
         attr_type = factory.box_attributes.get_attribute_type(name)
+
         if attr_type == Attribute.INTEGER:
             return int(value)
         if attr_type == Attribute.DOUBLE:
@@ -126,52 +186,157 @@ class TestbedInstance(testbed_impl.TestbedInstance):
     def action(self, time, guid, action):
         raise NotImplementedError
 
-    def trace(self, guid, trace_id):
-        fd = open("%s" % self.trace_filename(guid, trace_id), "r")
-        content = fd.read()
-        fd.close()
-        return content
-
-    def shutdown(self):
-        for element in self._elements.values():
-            element.destroy()
+    def trace_filepath(self, guid, trace_id):
+        filename = self._traces[guid][trace_id]
+        return os.path.join(self.home_directory, filename)
 
     def trace_filename(self, guid, trace_id):
-        # TODO: Need to be defined inside a home!!!! with and experiment id_code
-        filename = self._trace_filenames[guid][trace_id]
-        return os.path.join(self.home_directory, filename)
+        return self._traces[guid][trace_id]
 
     def follow_trace(self, guid, trace_id, filename):
-        if guid not in self._traces:
+        if not guid in self._traces:
             self._traces[guid] = dict()
         self._traces[guid][trace_id] = filename
 
-    def _load_ns3_module(self):
-        import ctypes
-        import imp
+    def shutdown(self):
+        for element in self._elements.itervalues():
+            if isinstance(element, self.LOCAL_TYPES):
+                # graceful shutdown of locally-implemented objects
+                element.cleanup()
+        if self.ns3:
+            if not self.ns3.Simulator.IsFinished():
+                self.stop()
+            
+            # TODO!!!! SHOULD WAIT UNTIL THE THREAD FINISHES
+            if self._simulator_thread:
+                self._simulator_thread.join()
+            
+            self.ns3.Simulator.Destroy()
+        
+        self._elements.clear()
+        
+        self._ns3 = None
+        sys.stdout.flush()
+        sys.stderr.flush()
+
+    def _simulator_run(self, condition):
+        # Run simulation
+        self.ns3.Simulator.Run()
+        # Signal condition on simulation end to notify waiting threads
+        condition.acquire()
+        condition.notifyAll()
+        condition.release()
+
+    def _schedule_event(self, condition, func, *args):
+        """Schedules event on running experiment"""
+        def execute_event(contextId, condition, has_event_occurred, func, *args):
+            # exec func
+            try:
+                func(*args)
+            finally:
+                # flag event occured
+                has_event_occurred[0] = True
+                # notify condition indicating attribute was set
+                condition.acquire()
+                condition.notifyAll()
+                condition.release()
+
+        # contextId is defined as general context
+        contextId = long(0xffffffff)
+        # delay 0 means that the event is expected to execute inmediately
+        delay = self.ns3.Seconds(0)
+        # flag to indicate that the event occured
+        # because bool is an inmutable object in python, in order to create a
+        # bool flag, a list is used as wrapper
+        has_event_occurred = [False]
+        condition.acquire()
+        try:
+            if not self.ns3.Simulator.IsFinished():
+                self.ns3.Simulator.ScheduleWithContext(contextId, delay, execute_event,
+                     condition, has_event_occurred, func, *args)
+                while not has_event_occurred[0] and not self.ns3.Simulator.IsFinished():
+                    condition.wait()
+        finally:
+            condition.release()
+
+    def _set_attribute(self, name, ns3_value, element):
+        if self.status() == TS.STATUS_STARTED:
+            # schedule the event in the Simulator
+            self._schedule_event(self._condition, self._set_ns3_attribute, 
+                    name, ns3_value, element)
+        else:
+            self._set_ns3_attribute(name, ns3_value, element)
+
+    def _get_attribute(self, name, ns3_value, element):
+        if self.status() == TS.STATUS_STARTED:
+            # schedule the event in the Simulator
+            self._schedule_event(self._condition, self._get_ns3_attribute, 
+                    name, ns3_value, element)
+        else:
+            self._get_ns3_attribute(name, ns3_value, element)
+
+    def _set_ns3_attribute(self, name, ns3_value, element):
+        element.SetAttribute(name, ns3_value)
+
+    def _get_ns3_attribute(self, name, ns3_value, element):
+        element.GetAttribute(name, ns3_value)
+
+    def _stop_simulation(self, time):
+        if self.status() == TS.STATUS_STARTED:
+            # schedule the event in the Simulator
+            self._schedule_event(self._condition, self._stop_ns3_simulation, 
+                    time)
+        else:
+            self._stop_ns3_simulation(time)
+
+    def _stop_ns3_simulation(self, time = TIME_NOW):
+        if not self.ns3:
+            return
+        if time == TIME_NOW:
+            self.ns3.Simulator.Stop()
+        else:
+            self.ns3.Simulator.Stop(self.ns3.Time(time))
 
-        bindings = self._attributes.get_attribute_value("ns3Bindings")
-        libfile = self._attributes.get_attribute_value("ns3Library")
+    def _to_ns3_value(self, guid, name, value):
+        factory_id = self._create[guid]
+        TypeId = self.ns3.TypeId()
+        typeid = TypeId.LookupByName(factory_id)
+        info = TypeId.AttributeInformation()
+        if not typeid.LookupAttributeByName(name, info):
+            raise RuntimeError("Attribute %s doesn't belong to element %s" \
+                   % (name, factory_id))
+        str_value = str(value)
+        if isinstance(value, bool):
+            str_value = str_value.lower()
+        checker = info.checker
+        ns3_value = checker.Create()
+        ns3_value.DeserializeFromString(str_value, checker)
+        return ns3_value
+
+    def _configure_ns3_module(self):
         simu_impl_type = self._attributes.get_attribute_value(
                 "SimulatorImplementationType")
+        sched_impl_type = self._attributes.get_attribute_value(
+                "SchedulerType")
         checksum = self._attributes.get_attribute_value("ChecksumEnabled")
+        stop_time = self._attributes.get_attribute_value("StopTime")
 
-        if libfile:
-            ctypes.CDLL(libfile, ctypes.RTLD_GLOBAL)
+        load_ns3_module()
 
-        path = [ os.path.dirname(__file__) ] + sys.path
-        if bindings:
-            path = [ bindings ] + path
-
-        module = imp.find_module ('ns3', path)
-        mod = imp.load_module ('ns3', *module)
-    
+        import ns3 as mod
         if simu_impl_type:
             value = mod.StringValue(simu_impl_type)
             mod.GlobalValue.Bind ("SimulatorImplementationType", value)
+        if sched_impl_type:
+            value = mod.StringValue(sched_impl_type)
+            mod.GlobalValue.Bind ("SchedulerType", value)
         if checksum:
             value = mod.BooleanValue(checksum)
             mod.GlobalValue.Bind ("ChecksumEnabled", value)
+        if stop_time:
+            value = mod.Time(stop_time)
+            mod.Simulator.Stop (value)
         return mod
 
     def _get_construct_parameters(self, guid):
@@ -180,10 +345,13 @@ class TestbedInstance(testbed_impl.TestbedInstance):
         factory_id = self._create[guid]
         TypeId = self.ns3.TypeId()
         typeid = TypeId.LookupByName(factory_id)
-        for name, value in params:
-            info = self.ns3.TypeId.AttributeInfo()
-            typeid.LookupAttributeByName(name, info)
-            if info.flags & TypeId.ATTR_CONSTRUCT == TypeId.ATTR_CONSTRUCT:
+        for name, value in params.iteritems():
+            info = self.ns3.TypeId.AttributeInformation()
+            found = typeid.LookupAttributeByName(name, info)
+            if found and \
+                (info.flags & TypeId.ATTR_CONSTRUCT == TypeId.ATTR_CONSTRUCT):
                 construct_params[name] = value
         return construct_params
 
+
+