src/nepi/testbeds/netns/execute.py

   1 #!/usr/bin/env python
   2 # -*- coding: utf-8 -*-
   3
   4 from constants import TESTBED_ID
   5 from nepi.core import testbed_impl
   6 from nepi.util.constants import TIME_NOW
   7 import os
   8 import fcntl
   9 import threading
  10
  11 class TestbedController(testbed_impl.TestbedController):
  12     from nepi.util.tunchannel_impl import TunChannel
  13
  14     class HostLock(object):
  15         # This class is used as a lock to prevent concurrency issues with more
  16         # than one instance of netns running in the same machine. Both in
  17         # different processes or different threads.
  18         taken = False
  19         processcond = threading.Condition()
  20
  21         def __init__(self, lockfile):
  22             processcond = self.__class__.processcond
  23
  24             processcond.acquire()
  25             try:
  26                 # It's not reentrant
  27                 while self.__class__.taken:
  28                     processcond.wait()
  29                 self.__class__.taken = True
  30             finally:
  31                 processcond.release()
  32
  33             self.lockfile = lockfile
  34             fcntl.flock(self.lockfile, fcntl.LOCK_EX)
  35
  36         def __del__(self):
  37             processcond = self.__class__.processcond
  38
  39             processcond.acquire()
  40             try:
  41                 assert self.__class__.taken, "HostLock unlocked without being locked!"
  42
  43                 fcntl.flock(self.lockfile, fcntl.LOCK_UN)
  44
  45                 # It's not reentrant
  46                 self.__class__.taken = False
  47                 processcond.notify()
  48             finally:
  49                 processcond.release()
  50
  51     def __init__(self, testbed_version):
  52         super(TestbedController, self).__init__(TESTBED_ID, testbed_version)
  53         self._netns = None
  54         self._home_directory = None
  55         self._traces = dict()
  56         self._netns_lock = open("/tmp/nepi-netns-lock","a")
  57
  58     def _lock(self):
  59         return self.HostLock(self._netns_lock)
  60
  61     @property
  62     def home_directory(self):
  63         return self._home_directory
  64
  65     @property
  66     def netns(self):
  67         return self._netns
  68
  69     def do_setup(self):
  70         lock = self._lock()
  71
  72         self._home_directory = self._attributes.\
  73             get_attribute_value("homeDirectory")
  74         # create home...
  75         home = os.path.normpath(self.home_directory)
  76         if not os.path.exists(home):
  77             os.makedirs(home, 0755)
  78
  79         self._netns = self._load_netns_module()
  80         super(TestbedController, self).do_setup()
  81
  82     def do_create(self):
  83         lock = self._lock()
  84         super(TestbedController, self).do_create()
  85
  86     def set(self, guid, name, value, time = TIME_NOW):
  87         super(TestbedController, self).set(guid, name, value, time)
  88         # TODO: take on account schedule time for the task
  89         factory_id = self._create[guid]
  90         factory = self._factories[factory_id]
  91         if factory.box_attributes.is_attribute_design_only(name):
  92             return
  93         element = self._elements.get(guid)
  94         if element:
  95             setattr(element, name, value)
  96
  97     def get(self, guid, name, time = TIME_NOW):
  98         value = super(TestbedController, self).get(guid, name, time)
  99         # TODO: take on account schedule time for the task
 100         factory_id = self._create[guid]
 101         factory = self._factories[factory_id]
 102         if factory.box_attributes.is_attribute_design_only(name):
 103             return value
 104         element = self._elements.get(guid)
 105         try:
 106             return getattr(element, name)
 107         except KeyError, AttributeError:
 108             return value
 109
 110     def action(self, time, guid, action):
 111         raise NotImplementedError
 112
 113     def shutdown(self):
 114         lock = self._lock()
 115
 116         for guid, traces in self._traces.iteritems():
 117             for trace_id, (trace, filename) in traces.iteritems():
 118                 if hasattr(trace, "close"):
 119                     trace.close()
 120         for guid, element in self._elements.iteritems():
 121             if isinstance(element, self.TunChannel):
 122                 element.Cleanup()
 123             else:
 124                 factory_id = self._create[guid]
 125                 if factory_id == "Node":
 126                     element.destroy()
 127         self._elements.clear()
 128
 129     def trace_filepath(self, guid, trace_id, filename = None):
 130         if not filename:
 131             (trace, filename) = self._traces[guid][trace_id]
 132         return os.path.join(self.home_directory, filename)
 133
 134     def follow_trace(self, guid, trace_id, trace, filename):
 135         if not guid in self._traces:
 136             self._traces[guid] = dict()
 137         self._traces[guid][trace_id] = (trace, filename)
 138
 139     def _load_netns_module(self):
 140         # TODO: Do something with the configuration!!!
 141         import sys
 142         __import__("netns")
 143         netns_mod = sys.modules["netns"]
 144         # enable debug
 145         enable_debug = self._attributes.get_attribute_value("enableDebug")
 146         if enable_debug:
 147             netns_mod.environ.set_log_level(netns_mod.environ.LOG_DEBUG)
 148         return netns_mod
 149