src/nepi/testbeds/netns/execute.py

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