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