src/nepi/core/execute.py

   1 #!/usr/bin/env python
   2 # -*- coding: utf-8 -*-
   3
   4 from nepi.core.attributes import Attribute, AttributesMap
   5 from nepi.core.connector import ConnectorTypeBase
   6 from nepi.util import validation
   7 from nepi.util.constants import STATUS_FINISHED, TIME_NOW
   8 from nepi.util.parser._xml import XmlExperimentParser
   9 import sys
  10 import re
  11 import threading
  12 import ConfigParser
  13 import os
  14 import collections
  15 import functools
  16
  17 ATTRIBUTE_PATTERN_BASE = re.compile(r"\{#\[(?P<label>[-a-zA-Z0-9._]*)\](?P<expr>(?P<component>\.addr\[[0-9]+\]|\.route\[[0-9]+\]|\.trace\[[0-9]+\])?.\[(?P<attribute>[-a-zA-Z0-9._]*)\])#}")
  18 ATTRIBUTE_PATTERN_GUID_SUB = r"{#[%(guid)s]%(expr)s#}"
  19 COMPONENT_PATTERN = re.compile(r"(?P<kind>[a-z]*)\[(?P<index>.*)\]")
  20
  21 class ConnectorType(ConnectorTypeBase):
  22     def __init__(self, testbed_id, factory_id, name, max = -1, min = 0):
  23         super(ConnectorType, self).__init__(testbed_id, factory_id, name, max, min)
  24         # from_connections -- connections where the other connector is the "From"
  25         # to_connections -- connections where the other connector is the "To"
  26         # keys in the dictionary correspond to the
  27         # connector_type_id for possible connections. The value is a tuple:
  28         # (can_cross, connect)
  29         # can_cross: indicates if the connection is allowed accros different
  30         #    testbed instances
  31         # code: is the connection function to be invoked when the elements
  32         #    are connected
  33         self._from_connections = dict()
  34         self._to_connections = dict()
  35
  36     def add_from_connection(self, testbed_id, factory_id, name, can_cross,
  37             init_code, compl_code):
  38         type_id = self.make_connector_type_id(testbed_id, factory_id, name)
  39         self._from_connections[type_id] = (can_cross, init_code, compl_code)
  40
  41     def add_to_connection(self, testbed_id, factory_id, name, can_cross,
  42             init_code, compl_code):
  43         type_id = self.make_connector_type_id(testbed_id, factory_id, name)
  44         self._to_connections[type_id] = (can_cross, init_code, compl_code)
  45
  46     def can_connect(self, testbed_id, factory_id, name, count,
  47             must_cross = False):
  48         connector_type_id = self.make_connector_type_id(testbed_id, factory_id, name)
  49         for lookup_type_id in self._type_resolution_order(connector_type_id):
  50             if lookup_type_id in self._from_connections:
  51                 (can_cross, init_code, compl_code) = self._from_connections[lookup_type_id]
  52             elif lookup_type_id in self._to_connections:
  53                 (can_cross, init_code, compl_code) = self._to_connections[lookup_type_id]
  54             else:
  55                 # keey trying
  56                 continue
  57             return not must_cross or can_cross
  58         else:
  59             return False
  60
  61     def _connect_to_code(self, testbed_id, factory_id, name):
  62         connector_type_id = self.make_connector_type_id(testbed_id, factory_id, name)
  63         for lookup_type_id in self._type_resolution_order(connector_type_id):
  64             if lookup_type_id in self._to_connections:
  65                 (can_cross, init_code, compl_code) = self._to_connections[lookup_type_id]
  66                 return (init_code, compl_code)
  67         else:
  68             return (False, False)
  69
  70     def connect_to_init_code(self, testbed_id, factory_id, name):
  71         return self._connect_to_code(testbed_id, factory_id, name)[0]
  72
  73     def connect_to_compl_code(self, testbed_id, factory_id, name):
  74         return self._connect_to_code(testbed_id, factory_id, name)[1]
  75
  76 class Factory(AttributesMap):
  77     def __init__(self, factory_id, create_function, start_function,
  78             stop_function, status_function,
  79             configure_function, preconfigure_function,
  80             allow_addresses = False, has_addresses = False,
  81             allow_routes = False, has_routes = False):
  82         super(Factory, self).__init__()
  83         self._factory_id = factory_id
  84         self._allow_addresses = bool(allow_addresses)
  85         self._allow_routes = bool(allow_routes)
  86         self._has_addresses = bool(has_addresses) or self._allow_addresses
  87         self._has_routes = bool(has_routes) or self._allow_routes
  88         self._create_function = create_function
  89         self._start_function = start_function
  90         self._stop_function = stop_function
  91         self._status_function = status_function
  92         self._configure_function = configure_function
  93         self._preconfigure_function = preconfigure_function
  94         self._connector_types = dict()
  95         self._traces = list()
  96         self._box_attributes = AttributesMap()
  97
  98     @property
  99     def factory_id(self):
 100         return self._factory_id
 101
 102     @property
 103     def allow_addresses(self):
 104         return self._allow_addresses
 105
 106     @property
 107     def allow_routes(self):
 108         return self._allow_routes
 109
 110     @property
 111     def has_addresses(self):
 112         return self._has_addresses
 113
 114     @property
 115     def has_routes(self):
 116         return self._has_routes
 117
 118     @property
 119     def box_attributes(self):
 120         return self._box_attributes
 121
 122     @property
 123     def create_function(self):
 124         return self._create_function
 125
 126     @property
 127     def start_function(self):
 128         return self._start_function
 129
 130     @property
 131     def stop_function(self):
 132         return self._stop_function
 133
 134     @property
 135     def status_function(self):
 136         return self._status_function
 137
 138     @property
 139     def configure_function(self):
 140         return self._configure_function
 141
 142     @property
 143     def preconfigure_function(self):
 144         return self._preconfigure_function
 145
 146     @property
 147     def traces(self):
 148         return self._traces
 149
 150     def connector_type(self, name):
 151         return self._connector_types[name]
 152
 153     def add_connector_type(self, connector_type):
 154         self._connector_types[connector_type.name] = connector_type
 155
 156     def add_trace(self, trace_id):
 157         self._traces.append(trace_id)
 158
 159     def add_box_attribute(self, name, help, type, value = None, range = None,
 160         allowed = None, flags = Attribute.NoFlags, validation_function = None):
 161         self._box_attributes.add_attribute(name, help, type, value, range,
 162                 allowed, flags, validation_function)
 163
 164 class TestbedController(object):
 165     def __init__(self, testbed_id, testbed_version):
 166         self._testbed_id = testbed_id
 167         self._testbed_version = testbed_version
 168
 169     @property
 170     def testbed_id(self):
 171         return self._testbed_id
 172
 173     @property
 174     def testbed_version(self):
 175         return self._testbed_version
 176
 177     @property
 178     def guids(self):
 179         raise NotImplementedError
 180
 181     def defer_configure(self, name, value):
 182         """Instructs setting a configuartion attribute for the testbed instance"""
 183         raise NotImplementedError
 184
 185     def defer_create(self, guid, factory_id):
 186         """Instructs creation of element """
 187         raise NotImplementedError
 188
 189     def defer_create_set(self, guid, name, value):
 190         """Instructs setting an initial attribute on an element"""
 191         raise NotImplementedError
 192
 193     def defer_factory_set(self, guid, name, value):
 194         """Instructs setting an attribute on a factory"""
 195         raise NotImplementedError
 196
 197     def defer_connect(self, guid1, connector_type_name1, guid2,
 198             connector_type_name2):
 199         """Instructs creation of a connection between the given connectors"""
 200         raise NotImplementedError
 201
 202     def defer_cross_connect(self,
 203             guid, connector_type_name,
 204             cross_guid, cross_testbed_guid,
 205             cross_testbed_id, cross_factory_id,
 206             cross_connector_type_name):
 207         """
 208         Instructs creation of a connection between the given connectors
 209         of different testbed instances
 210         """
 211         raise NotImplementedError
 212
 213     def defer_add_trace(self, guid, trace_id):
 214         """Instructs the addition of a trace"""
 215         raise NotImplementedError
 216
 217     def defer_add_address(self, guid, address, netprefix, broadcast):
 218         """Instructs the addition of an address"""
 219         raise NotImplementedError
 220
 221     def defer_add_route(self, guid, destination, netprefix, nexthop):
 222         """Instructs the addition of a route"""
 223         raise NotImplementedError
 224
 225     def do_setup(self):
 226         """After do_setup the testbed initial configuration is done"""
 227         raise NotImplementedError
 228
 229     def do_create(self):
 230         """
 231         After do_create all instructed elements are created and
 232         attributes setted
 233         """
 234         raise NotImplementedError
 235
 236     def do_connect_init(self):
 237         """
 238         After do_connect_init all internal connections between testbed elements
 239         are initiated
 240         """
 241         raise NotImplementedError
 242
 243     def do_connect_compl(self):
 244         """
 245         After do_connect all internal connections between testbed elements
 246         are completed
 247         """
 248         raise NotImplementedError
 249
 250     def do_preconfigure(self):
 251         """
 252         Done just before resolving netrefs, after connection, before cross connections,
 253         useful for early stages of configuration, for setting up stuff that might be
 254         required for netref resolution.
 255         """
 256         raise NotImplementedError
 257
 258     def do_configure(self):
 259         """After do_configure elements are configured"""
 260         raise NotImplementedError
 261
 262     def do_cross_connect_init(self, cross_data):
 263         """
 264         After do_cross_connect_init initiation of all external connections
 265         between different testbed elements is performed
 266         """
 267         raise NotImplementedError
 268
 269     def do_cross_connect_compl(self, cross_data):
 270         """
 271         After do_cross_connect_compl completion of all external connections
 272         between different testbed elements is performed
 273         """
 274         raise NotImplementedError
 275
 276     def start(self):
 277         raise NotImplementedError
 278
 279     def stop(self):
 280         raise NotImplementedError
 281
 282     def set(self, guid, name, value, time = TIME_NOW):
 283         raise NotImplementedError
 284
 285     def get(self, guid, name, time = TIME_NOW):
 286         raise NotImplementedError
 287
 288     def get_route(self, guid, index, attribute):
 289         """
 290         Params:
 291
 292             guid: guid of box to query
 293             index: number of routing entry to fetch
 294             attribute: one of Destination, NextHop, NetPrefix
 295         """
 296         raise NotImplementedError
 297
 298     def get_address(self, guid, index, attribute='Address'):
 299         """
 300         Params:
 301
 302             guid: guid of box to query
 303             index: number of inteface to select
 304             attribute: one of Address, NetPrefix, Broadcast
 305         """
 306         raise NotImplementedError
 307
 308     def get_attribute_list(self, guid):
 309         raise NotImplementedError
 310
 311     def action(self, time, guid, action):
 312         raise NotImplementedError
 313
 314     def status(self, guid):
 315         raise NotImplementedError
 316
 317     def trace(self, guid, trace_id, attribute='value'):
 318         raise NotImplementedError
 319
 320     def shutdown(self):
 321         raise NotImplementedError
 322
 323 class ExperimentController(object):
 324     def __init__(self, experiment_xml, root_dir):
 325         self._experiment_xml = experiment_xml
 326         self._testbeds = dict()
 327         self._deployment_config = dict()
 328         self._netrefs = collections.defaultdict(set)
 329         self._testbed_netrefs = collections.defaultdict(set)
 330         self._cross_data = dict()
 331         self._root_dir = root_dir
 332         self._netreffed_testbeds = set()
 333         self._guids_in_testbed_cache = dict()
 334
 335         self.persist_experiment_xml()
 336
 337     @property
 338     def experiment_xml(self):
 339         return self._experiment_xml
 340
 341     def persist_experiment_xml(self):
 342         xml_path = os.path.join(self._root_dir, "experiment.xml")
 343         f = open(xml_path, "w")
 344         f.write(self._experiment_xml)
 345         f.close()
 346
 347     def trace(self, testbed_guid, guid, trace_id, attribute='value'):
 348         return self._testbeds[testbed_guid].trace(guid, trace_id, attribute)
 349
 350     @staticmethod
 351     def _parallel(callables):
 352         excs = []
 353         def wrap(callable):
 354             @functools.wraps(callable)
 355             def wrapped(*p, **kw):
 356                 try:
 357                     callable(*p, **kw)
 358                 except Exception,e:
 359                     import traceback
 360                     traceback.print_exc(file=sys.stderr)
 361                     excs.append(e)
 362             return wrapped
 363         threads = [ threading.Thread(target=wrap(callable)) for callable in callables ]
 364         for thread in threads:
 365             thread.start()
 366         for thread in threads:
 367             thread.join()
 368         for exc in excs:
 369             raise exc
 370
 371     def start(self):
 372         parser = XmlExperimentParser()
 373         data = parser.from_xml_to_data(self._experiment_xml)
 374
 375         self._init_testbed_controllers(data)
 376
 377         # persist testbed connection data, for potential recovery
 378         self._persist_testbed_proxies()
 379
 380         def steps_to_configure(self, allowed_guids):
 381             # perform setup in parallel for all test beds,
 382             # wait for all threads to finish
 383             self._parallel([testbed.do_setup
 384                             for guid,testbed in self._testbeds.iteritems()
 385                             if guid in allowed_guids])
 386
 387             # perform create-connect in parallel, wait
 388             # (internal connections only)
 389             self._parallel([testbed.do_create
 390                             for guid,testbed in self._testbeds.iteritems()
 391                             if guid in allowed_guids])
 392
 393             self._parallel([testbed.do_connect_init
 394                             for guid,testbed in self._testbeds.iteritems()
 395                             if guid in allowed_guids])
 396
 397             self._parallel([testbed.do_connect_compl
 398                             for guid,testbed in self._testbeds.iteritems()
 399                             if guid in allowed_guids])
 400
 401             self._parallel([testbed.do_preconfigure
 402                             for guid,testbed in self._testbeds.iteritems()
 403                             if guid in allowed_guids])
 404
 405         steps_to_configure(self, self._testbeds)
 406
 407         if self._netreffed_testbeds:
 408             # initally resolve netrefs
 409             self.do_netrefs(data, fail_if_undefined=False)
 410
 411             # rinse and repeat, for netreffed testbeds
 412             netreffed_testbeds = set(self._netreffed_testbeds)
 413
 414             self._init_testbed_controllers(data)
 415
 416             # persist testbed connection data, for potential recovery
 417             self._persist_testbed_proxies()
 418
 419             # configure dependant testbeds
 420             steps_to_configure(self, netreffed_testbeds)
 421
 422         # final netref step, fail if anything's left unresolved
 423         self.do_netrefs(data, fail_if_undefined=True)
 424
 425         self._program_testbed_cross_connections(data)
 426
 427         # perform do_configure in parallel for al testbeds
 428         # (it's internal configuration for each)
 429         self._parallel([testbed.do_configure
 430                         for testbed in self._testbeds.itervalues()])
 431
 432
 433         #print >>sys.stderr, "DO IT"
 434         #import time
 435         #time.sleep(60)
 436
 437         # cross-connect (cannot be done in parallel)
 438         for guid, testbed in self._testbeds.iteritems():
 439             cross_data = self._get_cross_data(guid)
 440             testbed.do_cross_connect_init(cross_data)
 441         for guid, testbed in self._testbeds.iteritems():
 442             cross_data = self._get_cross_data(guid)
 443             testbed.do_cross_connect_compl(cross_data)
 444
 445         # start experiment (parallel start on all testbeds)
 446         self._parallel([testbed.start
 447                         for testbed in self._testbeds.itervalues()])
 448
 449     def _persist_testbed_proxies(self):
 450         TRANSIENT = ('Recover',)
 451
 452         # persist access configuration for all testbeds, so that
 453         # recovery mode can reconnect to them if it becomes necessary
 454         conf = ConfigParser.RawConfigParser()
 455         for testbed_guid, testbed_config in self._deployment_config.iteritems():
 456             testbed_guid = str(testbed_guid)
 457             conf.add_section(testbed_guid)
 458             for attr in testbed_config.attributes_list:
 459                 if attr not in TRANSIENT:
 460                     conf.set(testbed_guid, attr,
 461                         testbed_config.get_attribute_value(attr))
 462
 463         f = open(os.path.join(self._root_dir, 'deployment_config.ini'), 'w')
 464         conf.write(f)
 465         f.close()
 466
 467     def _load_testbed_proxies(self):
 468         TYPEMAP = {
 469             STRING : 'get',
 470             INTEGER : 'getint',
 471             FLOAT : 'getfloat',
 472             BOOLEAN : 'getboolean',
 473         }
 474
 475         # deferred import because proxy needs
 476         # our class definitions to define proxies
 477         import nepi.util.proxy as proxy
 478
 479         conf = ConfigParser.RawConfigParser()
 480         conf.read(os.path.join(self._root_dir, 'deployment_config.ini'))
 481         for testbed_guid in conf.sections():
 482             testbed_config = proxy.AccessConfiguration()
 483             for attr in conf.options(testbed_guid):
 484                 testbed_config.set_attribute_value(attr,
 485                     conf.get(testbed_guid, attr) )
 486
 487             testbed_guid = str(testbed_guid)
 488             conf.add_section(testbed_guid)
 489             for attr in testbed_config.attributes_list:
 490                 if attr not in TRANSIENT:
 491                     getter = getattr(conf, TYPEMAP.get(
 492                         testbed_config.get_attribute_type(attr),
 493                         'get') )
 494                     testbed_config.set_attribute_value(
 495                         testbed_guid, attr, getter(attr))
 496
 497     def _unpersist_testbed_proxies(self):
 498         try:
 499             os.remove(os.path.join(self._root_dir, 'deployment_config.ini'))
 500         except:
 501             # Just print exceptions, this is just cleanup
 502             import traceback
 503             traceback.print_exc(file=sys.stderr)
 504
 505     def stop(self):
 506        for testbed in self._testbeds.values():
 507            testbed.stop()
 508        self._unpersist_testbed_proxies()
 509
 510     def recover(self):
 511         # reload perviously persisted testbed access configurations
 512         self._load_testbed_proxies()
 513
 514         # recreate testbed proxies by reconnecting only
 515         self._init_testbed_controllers(recover = True)
 516
 517         # another time, for netrefs
 518         self._init_testbed_controllers(recover = True)
 519
 520     def is_finished(self, guid):
 521         for testbed in self._testbeds.values():
 522             for guid_ in testbed.guids:
 523                 if guid_ == guid:
 524                     return testbed.status(guid) == STATUS_FINISHED
 525         raise RuntimeError("No element exists with guid %d" % guid)
 526
 527     def set(self, testbed_guid, guid, name, value, time = TIME_NOW):
 528         testbed = self._testbeds[testbed_guid]
 529         testbed.set(guid, name, value, time)
 530
 531     def get(self, testbed_guid, guid, name, time = TIME_NOW):
 532         testbed = self._testbeds[testbed_guid]
 533         return testbed.get(guid, name, time)
 534
 535     def shutdown(self):
 536         for testbed in self._testbeds.values():
 537             testbed.shutdown()
 538
 539     def _guids_in_testbed(self, testbed_guid):
 540         if testbed_guid not in self._testbeds:
 541             return set()
 542         if testbed_guid not in self._guids_in_testbed_cache:
 543             self._guids_in_testbed_cache[testbed_guid] = \
 544                 set(self._testbeds[testbed_guid].guids)
 545         return self._guids_in_testbed_cache[testbed_guid]
 546
 547     @staticmethod
 548     def _netref_component_split(component):
 549         match = COMPONENT_PATTERN.match(component)
 550         if match:
 551             return match.group("kind"), match.group("index")
 552         else:
 553             return component, None
 554
 555     _NETREF_COMPONENT_GETTERS = {
 556         'addr':
 557             lambda testbed, guid, index, name:
 558                 testbed.get_address(guid, int(index), name),
 559         'route' :
 560             lambda testbed, guid, index, name:
 561                 testbed.get_route(guid, int(index), name),
 562         'trace' :
 563             lambda testbed, guid, index, name:
 564                 testbed.trace(guid, index, name),
 565         '' :
 566             lambda testbed, guid, index, name:
 567                 testbed.get(guid, name),
 568     }
 569
 570     def resolve_netref_value(self, value):
 571         match = ATTRIBUTE_PATTERN_BASE.search(value)
 572         if match:
 573             label = match.group("label")
 574             if label.startswith('GUID-'):
 575                 ref_guid = int(label[5:])
 576                 if ref_guid:
 577                     expr = match.group("expr")
 578                     component = (match.group("component") or "")[1:] # skip the dot
 579                     attribute = match.group("attribute")
 580
 581                     # split compound components into component kind and index
 582                     # eg: 'addr[0]' -> ('addr', '0')
 583                     component, component_index = self._netref_component_split(component)
 584
 585                     # find object and resolve expression
 586                     for ref_testbed_guid, ref_testbed in self._testbeds.iteritems():
 587                         if component not in self._NETREF_COMPONENT_GETTERS:
 588                             raise ValueError, "Malformed netref: %r - unknown component" % (expr,)
 589                         elif ref_guid not in self._guids_in_testbed(ref_testbed_guid):
 590                             pass
 591                         else:
 592                             ref_value = self._NETREF_COMPONENT_GETTERS[component](
 593                                 ref_testbed, ref_guid, component_index, attribute)
 594                             if ref_value:
 595                                 return value.replace(match.group(), ref_value)
 596         # couldn't find value
 597         return None
 598
 599     def do_netrefs(self, data, fail_if_undefined = False):
 600         # element netrefs
 601         for (testbed_guid, guid), attrs in self._netrefs.items():
 602             testbed = self._testbeds[testbed_guid]
 603             allreplaced = True
 604             for name in attrs:
 605                 value = testbed.get(guid, name)
 606                 if isinstance(value, basestring):
 607                     ref_value = self.resolve_netref_value(value)
 608                     if ref_value is not None:
 609                         testbed.set(guid, name, ref_value)
 610                     elif fail_if_undefined:
 611                         raise ValueError, "Unresolvable netref in: %r" % (value,)
 612                     else:
 613                         allreplaced = False
 614             if allreplaced:
 615                 del self._netrefs[(testbed_guid, guid)]
 616
 617         # testbed netrefs
 618         for testbed_guid, attrs in self._testbed_netrefs.items():
 619             tb_data = dict(data.get_attribute_data(testbed_guid))
 620             if data:
 621                 allreplaced = True
 622                 for name in attrs:
 623                     value = tb_data.get(name)
 624                     if isinstance(value, basestring):
 625                         ref_value = self.resolve_netref_value(value)
 626                         if ref_value is not None:
 627                             data.set_attribute_data(testbed_guid, name, ref_value)
 628                         elif fail_if_undefined:
 629                             raise ValueError, "Unresolvable netref in: %r" % (value,)
 630                         else:
 631                             allreplaced = False
 632                 if allreplaced:
 633                     del self._testbed_netrefs[testbed_guid]
 634
 635
 636     def _init_testbed_controllers(self, data, recover = False):
 637         blacklist_testbeds = set(self._testbeds)
 638         element_guids = list()
 639         label_guids = dict()
 640         data_guids = data.guids
 641
 642         # create testbed controllers
 643         for guid in data_guids:
 644             if data.is_testbed_data(guid):
 645                 if guid not in self._testbeds:
 646                     self._create_testbed_controller(guid, data, element_guids,
 647                             recover)
 648             else:
 649                 (testbed_guid, factory_id) = data.get_box_data(guid)
 650                 if testbed_guid not in blacklist_testbeds:
 651                     element_guids.append(guid)
 652                     label = data.get_attribute_data(guid, "label")
 653                     if label is not None:
 654                         if label in label_guids:
 655                             raise RuntimeError, "Label %r is not unique" % (label,)
 656                         label_guids[label] = guid
 657
 658         # replace references to elements labels for its guid
 659         self._resolve_labels(data, data_guids, label_guids)
 660
 661         # program testbed controllers
 662         if not recover:
 663             self._program_testbed_controllers(element_guids, data)
 664
 665     def _resolve_labels(self, data, data_guids, label_guids):
 666         netrefs = self._netrefs
 667         testbed_netrefs = self._testbed_netrefs
 668         for guid in data_guids:
 669             for name, value in data.get_attribute_data(guid):
 670                 if isinstance(value, basestring):
 671                     match = ATTRIBUTE_PATTERN_BASE.search(value)
 672                     if match:
 673                         label = match.group("label")
 674                         if not label.startswith('GUID-'):
 675                             ref_guid = label_guids.get(label)
 676                             if ref_guid is not None:
 677                                 value = ATTRIBUTE_PATTERN_BASE.sub(
 678                                     ATTRIBUTE_PATTERN_GUID_SUB % dict(
 679                                         guid = 'GUID-%d' % (ref_guid,),
 680                                         expr = match.group("expr"),
 681                                         label = label),
 682                                     value)
 683                                 data.set_attribute_data(guid, name, value)
 684
 685                                 # memorize which guid-attribute pairs require
 686                                 # postprocessing, to avoid excessive controller-testbed
 687                                 # communication at configuration time
 688                                 # (which could require high-latency network I/O)
 689                                 if not data.is_testbed_data(guid):
 690                                     (testbed_guid, factory_id) = data.get_box_data(guid)
 691                                     netrefs[(testbed_guid, guid)].add(name)
 692                                 else:
 693                                     testbed_netrefs[guid].add(name)
 694
 695     def _create_testbed_controller(self, guid, data, element_guids, recover):
 696         (testbed_id, testbed_version) = data.get_testbed_data(guid)
 697         deployment_config = self._deployment_config.get(guid)
 698
 699         # deferred import because proxy needs
 700         # our class definitions to define proxies
 701         import nepi.util.proxy as proxy
 702
 703         if deployment_config is None:
 704             # need to create one
 705             deployment_config = proxy.AccessConfiguration()
 706
 707             for (name, value) in data.get_attribute_data(guid):
 708                 if value is not None and deployment_config.has_attribute(name):
 709                     # if any deployment config attribute has a netref, we can't
 710                     # create this controller yet
 711                     if isinstance(value, basestring) and ATTRIBUTE_PATTERN_BASE.search(value):
 712                         # remember to re-issue this one
 713                         self._netreffed_testbeds.add(guid)
 714                         return
 715
 716                     # copy deployment config attribute
 717                     deployment_config.set_attribute_value(name, value)
 718
 719             # commit config
 720             self._deployment_config[guid] = deployment_config
 721
 722         if deployment_config is not None:
 723             # force recovery mode
 724             deployment_config.set_attribute_value("recover",recover)
 725
 726         testbed = proxy.create_testbed_controller(testbed_id,
 727                 testbed_version, deployment_config)
 728         for (name, value) in data.get_attribute_data(guid):
 729             testbed.defer_configure(name, value)
 730         self._testbeds[guid] = testbed
 731         if guid in self._netreffed_testbeds:
 732             self._netreffed_testbeds.remove(guid)
 733
 734     def _program_testbed_controllers(self, element_guids, data):
 735         for guid in element_guids:
 736             (testbed_guid, factory_id) = data.get_box_data(guid)
 737             testbed = self._testbeds.get(testbed_guid)
 738             if testbed:
 739                 testbed.defer_create(guid, factory_id)
 740                 for (name, value) in data.get_attribute_data(guid):
 741                     testbed.defer_create_set(guid, name, value)
 742
 743         for guid in element_guids:
 744             (testbed_guid, factory_id) = data.get_box_data(guid)
 745             testbed = self._testbeds.get(testbed_guid)
 746             if testbed:
 747                 for (connector_type_name, cross_guid, cross_connector_type_name) \
 748                         in data.get_connection_data(guid):
 749                     (testbed_guid, factory_id) = data.get_box_data(guid)
 750                     (cross_testbed_guid, cross_factory_id) = data.get_box_data(
 751                             cross_guid)
 752                     if testbed_guid == cross_testbed_guid:
 753                         testbed.defer_connect(guid, connector_type_name,
 754                                 cross_guid, cross_connector_type_name)
 755                 for trace_id in data.get_trace_data(guid):
 756                     testbed.defer_add_trace(guid, trace_id)
 757                 for (autoconf, address, netprefix, broadcast) in \
 758                         data.get_address_data(guid):
 759                     if address != None:
 760                         testbed.defer_add_address(guid, address, netprefix,
 761                                 broadcast)
 762                 for (destination, netprefix, nexthop) in data.get_route_data(guid):
 763                     testbed.defer_add_route(guid, destination, netprefix, nexthop)
 764
 765     def _program_testbed_cross_connections(self, data):
 766         data_guids = data.guids
 767
 768         for guid in data_guids:
 769             if not data.is_testbed_data(guid):
 770                 (testbed_guid, factory_id) = data.get_box_data(guid)
 771                 testbed = self._testbeds.get(testbed_guid)
 772                 if testbed:
 773                     for (connector_type_name, cross_guid, cross_connector_type_name) \
 774                             in data.get_connection_data(guid):
 775                         (testbed_guid, factory_id) = data.get_box_data(guid)
 776                         (cross_testbed_guid, cross_factory_id) = data.get_box_data(
 777                                 cross_guid)
 778                         if testbed_guid != cross_testbed_guid:
 779                             cross_testbed = self._testbeds[cross_testbed_guid]
 780                             cross_testbed_id = cross_testbed.testbed_id
 781                             testbed.defer_cross_connect(guid, connector_type_name, cross_guid,
 782                                     cross_testbed_guid, cross_testbed_id, cross_factory_id,
 783                                     cross_connector_type_name)
 784                             # save cross data for later
 785                             self._add_crossdata(testbed_guid, guid, cross_testbed_guid,
 786                                     cross_guid)
 787
 788     def _add_crossdata(self, testbed_guid, guid, cross_testbed_guid, cross_guid):
 789         if testbed_guid not in self._cross_data:
 790             self._cross_data[testbed_guid] = dict()
 791         if cross_testbed_guid not in self._cross_data[testbed_guid]:
 792             self._cross_data[testbed_guid][cross_testbed_guid] = set()
 793         self._cross_data[testbed_guid][cross_testbed_guid].add(cross_guid)
 794
 795     def _get_cross_data(self, testbed_guid):
 796         cross_data = dict()
 797         if not testbed_guid in self._cross_data:
 798             return cross_data
 799         for cross_testbed_guid, guid_list in \
 800                 self._cross_data[testbed_guid].iteritems():
 801             cross_data[cross_testbed_guid] = dict()
 802             cross_testbed = self._testbeds[cross_testbed_guid]
 803             for cross_guid in guid_list:
 804                 elem_cross_data = dict(
 805                     _guid = cross_guid,
 806                     _testbed_guid = cross_testbed_guid,
 807                     _testbed_id = cross_testbed.testbed_id,
 808                     _testbed_version = cross_testbed.testbed_version)
 809                 cross_data[cross_testbed_guid][cross_guid] = elem_cross_data
 810                 attributes_list = cross_testbed.get_attribute_list(cross_guid)
 811                 for attr_name in attributes_list:
 812                     attr_value = cross_testbed.get(cross_guid, attr_name)
 813                     elem_cross_data[attr_name] = attr_value
 814         return cross_data
 815