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, exclude = False):
 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_design_xml = experiment_xml
 202         self._experiment_execute_xml = None
 203         self._testbeds = dict()
 204         self._deployment_config = dict()
 205         self._netrefs = collections.defaultdict(set)
 206         self._testbed_netrefs = collections.defaultdict(set)
 207         self._cross_data = dict()
 208         self._root_dir = root_dir
 209         self._netreffed_testbeds = set()
 210         self._guids_in_testbed_cache = dict()
 211
 212         self.persist_experiment_xml()
 213
 214     @property
 215     def experiment_design_xml(self):
 216         return self._experiment_design_xml
 217
 218     @property
 219     def experiment_execute_xml(self):
 220         return self._experiment_execute_xml
 221
 222     @property
 223     def guids(self):
 224         guids = list()
 225         for testbed_guid in self._testbeds.keys():
 226             _guids = self._guids_in_testbed(testbed_guid)
 227             if _guids:
 228                 guids.extend(_guids)
 229         return guids
 230
 231     def persist_experiment_xml(self):
 232         xml_path = os.path.join(self._root_dir, "experiment-design.xml")
 233         f = open(xml_path, "w")
 234         f.write(self._experiment_design_xml)
 235         f.close()
 236
 237     def persist_execute_xml(self):
 238         xml_path = os.path.join(self._root_dir, "experiment-execute.xml")
 239         f = open(xml_path, "w")
 240         f.write(self._experiment_execute_xml)
 241         f.close()
 242
 243     def trace(self, guid, trace_id, attribute='value'):
 244         testbed = self._testbed_for_guid(guid)
 245         if testbed != None:
 246             return testbed.trace(guid, trace_id, attribute)
 247         raise RuntimeError("No element exists with guid %d" % guid)
 248
 249     def traces_info(self):
 250         traces_info = dict()
 251         for guid, testbed in self._testbeds.iteritems():
 252             tinfo = testbed.traces_info()
 253             if tinfo:
 254                 traces_info[guid] = testbed.traces_info()
 255         return traces_info
 256
 257     @staticmethod
 258     def _parallel(callables):
 259         excs = []
 260         def wrap(callable):
 261             @functools.wraps(callable)
 262             def wrapped(*p, **kw):
 263                 try:
 264                     callable(*p, **kw)
 265                 except:
 266                     import traceback
 267                     traceback.print_exc(file=sys.stderr)
 268                     excs.append(sys.exc_info())
 269             return wrapped
 270         threads = [ threading.Thread(target=wrap(callable)) for callable in callables ]
 271         for thread in threads:
 272             thread.start()
 273         for thread in threads:
 274             thread.join()
 275         for exc in excs:
 276             eTyp, eVal, eLoc = exc
 277             raise eTyp, eVal, eLoc
 278
 279     def start(self):
 280         parser = XmlExperimentParser()
 281         data = parser.from_xml_to_data(self._experiment_design_xml)
 282
 283         # instantiate testbed controllers
 284         self._init_testbed_controllers(data)
 285
 286         # persist testbed connection data, for potential recovery
 287         self._persist_testbed_proxies()
 288
 289         def steps_to_configure(self, allowed_guids):
 290             # perform setup in parallel for all test beds,
 291             # wait for all threads to finish
 292             self._parallel([testbed.do_setup
 293                             for guid,testbed in self._testbeds.iteritems()
 294                             if guid in allowed_guids])
 295
 296             # perform create-connect in parallel, wait
 297             # (internal connections only)
 298             self._parallel([testbed.do_create
 299                             for guid,testbed in self._testbeds.iteritems()
 300                             if guid in allowed_guids])
 301
 302             self._parallel([testbed.do_connect_init
 303                             for guid,testbed in self._testbeds.iteritems()
 304                             if guid in allowed_guids])
 305
 306             self._parallel([testbed.do_connect_compl
 307                             for guid,testbed in self._testbeds.iteritems()
 308                             if guid in allowed_guids])
 309
 310             self._parallel([testbed.do_preconfigure
 311                             for guid,testbed in self._testbeds.iteritems()
 312                             if guid in allowed_guids])
 313             self._clear_caches()
 314
 315         steps_to_configure(self, self._testbeds)
 316
 317         if self._netreffed_testbeds:
 318             # initally resolve netrefs
 319             self.do_netrefs(data, fail_if_undefined=False)
 320
 321             # rinse and repeat, for netreffed testbeds
 322             netreffed_testbeds = set(self._netreffed_testbeds)
 323
 324             self._init_testbed_controllers(data)
 325
 326             # persist testbed connection data, for potential recovery
 327             self._persist_testbed_proxies()
 328
 329             # configure dependant testbeds
 330             steps_to_configure(self, netreffed_testbeds)
 331
 332         # final netref step, fail if anything's left unresolved
 333         self.do_netrefs(data, fail_if_undefined=True)
 334
 335         # perform do_configure in parallel for al testbeds
 336         # (it's internal configuration for each)
 337         self._parallel([testbed.do_configure
 338                         for testbed in self._testbeds.itervalues()])
 339
 340         self._clear_caches()
 341
 342         #print >>sys.stderr, "DO IT"
 343         #import time
 344         #time.sleep(60)
 345
 346         # cross-connect (cannot be done in parallel)
 347         for guid, testbed in self._testbeds.iteritems():
 348             cross_data = self._get_cross_data(guid)
 349             testbed.do_cross_connect_init(cross_data)
 350         for guid, testbed in self._testbeds.iteritems():
 351             cross_data = self._get_cross_data(guid)
 352             testbed.do_cross_connect_compl(cross_data)
 353
 354         self._clear_caches()
 355
 356         # Last chance to configure (parallel on all testbeds)
 357         self._parallel([testbed.do_prestart
 358                         for testbed in self._testbeds.itervalues()])
 359
 360         self._clear_caches()
 361
 362         # update execution xml with execution-specific values
 363         # TODO: BUG! BUggy code! cant stand all serializing all attribute values (ej: tun_key which is non ascci)"
 364         self._update_execute_xml()
 365         self.persist_execute_xml()
 366
 367         # start experiment (parallel start on all testbeds)
 368         self._parallel([testbed.start
 369                         for testbed in self._testbeds.itervalues()])
 370
 371         self._clear_caches()
 372
 373     def _clear_caches(self):
 374         # Cleaning cache for safety.
 375         self._guids_in_testbed_cache = dict()
 376
 377     def _persist_testbed_proxies(self):
 378         TRANSIENT = ('Recover',)
 379
 380         # persist access configuration for all testbeds, so that
 381         # recovery mode can reconnect to them if it becomes necessary
 382         conf = ConfigParser.RawConfigParser()
 383         for testbed_guid, testbed_config in self._deployment_config.iteritems():
 384             testbed_guid = str(testbed_guid)
 385             conf.add_section(testbed_guid)
 386             for attr in testbed_config.get_attribute_list():
 387                 if attr not in TRANSIENT:
 388                     conf.set(testbed_guid, attr,
 389                         testbed_config.get_attribute_value(attr))
 390
 391         f = open(os.path.join(self._root_dir, 'deployment_config.ini'), 'w')
 392         conf.write(f)
 393         f.close()
 394
 395     def _load_testbed_proxies(self):
 396         TYPEMAP = {
 397             STRING : 'get',
 398             INTEGER : 'getint',
 399             FLOAT : 'getfloat',
 400             BOOLEAN : 'getboolean',
 401         }
 402
 403         # deferred import because proxy needs
 404         # our class definitions to define proxies
 405         import nepi.util.proxy as proxy
 406
 407         conf = ConfigParser.RawConfigParser()
 408         conf.read(os.path.join(self._root_dir, 'deployment_config.ini'))
 409         for testbed_guid in conf.sections():
 410             testbed_config = proxy.AccessConfiguration()
 411             for attr in conf.options(testbed_guid):
 412                 testbed_config.set_attribute_value(attr,
 413                     conf.get(testbed_guid, attr) )
 414
 415             testbed_guid = str(testbed_guid)
 416             conf.add_section(testbed_guid)
 417             for attr in testbed_config.get_attribute_list():
 418                 if attr not in TRANSIENT:
 419                     getter = getattr(conf, TYPEMAP.get(
 420                         testbed_config.get_attribute_type(attr),
 421                         'get') )
 422                     testbed_config.set_attribute_value(
 423                         testbed_guid, attr, getter(attr))
 424
 425     def _unpersist_testbed_proxies(self):
 426         try:
 427             os.remove(os.path.join(self._root_dir, 'deployment_config.ini'))
 428         except:
 429             # Just print exceptions, this is just cleanup
 430             import traceback
 431             ######## BUG ##########
 432             #BUG: If the next line is uncomented pyQt explodes when shutting down the experiment !!!!!!!!
 433             #traceback.print_exc(file=sys.stderr)
 434
 435     def _update_execute_xml(self):
 436         # For all testbeds,
 437         #   For all elements in testbed,
 438         #       - gather immutable execute-readable attribuets lists
 439         #         asynchronously
 440         # Generate new design description from design xml
 441         # (Wait for attributes lists - implicit syncpoint)
 442         # For all testbeds,
 443         #   For all elements in testbed,
 444         #       - gather all immutable execute-readable attribute
 445         #         values, asynchronously
 446         # (Wait for attribute values - implicit syncpoint)
 447         # For all testbeds,
 448         #   For all elements in testbed,
 449         #       - inject non-None values into new design
 450         # Generate execute xml from new design
 451
 452         def undefer(deferred):
 453             if hasattr(deferred, '_get'):
 454                 return deferred._get()
 455             else:
 456                 return deferred
 457
 458         attribute_lists = dict(
 459             (testbed_guid, collections.defaultdict(dict))
 460             for testbed_guid in self._testbeds
 461         )
 462
 463         for testbed_guid, testbed in self._testbeds.iteritems():
 464             guids = self._guids_in_testbed(testbed_guid)
 465             for guid in guids:
 466                 attribute_lists[testbed_guid][guid] = \
 467                     testbed.get_attribute_list_deferred(guid, Attribute.ExecImmutable)
 468
 469         parser = XmlExperimentParser()
 470         execute_data = parser.from_xml_to_data(self._experiment_design_xml)
 471
 472         attribute_values = dict(
 473             (testbed_guid, collections.defaultdict(dict))
 474             for testbed_guid in self._testbeds
 475         )
 476
 477         for testbed_guid, testbed_attribute_lists in attribute_lists.iteritems():
 478             testbed = self._testbeds[testbed_guid]
 479             for guid, attribute_list in testbed_attribute_lists.iteritems():
 480                 attribute_list = undefer(attribute_list)
 481                 attribute_values[testbed_guid][guid] = dict(
 482                     (attribute, testbed.get_deferred(guid, attribute))
 483                     for attribute in attribute_list
 484                 )
 485
 486         for testbed_guid, testbed_attribute_values in attribute_values.iteritems():
 487             for guid, attribute_values in testbed_attribute_values.iteritems():
 488                 for attribute, value in attribute_values.iteritems():
 489                     value = undefer(value)
 490                     if value is not None:
 491                         execute_data.add_attribute_data(guid, attribute, value)
 492
 493         self._experiment_execute_xml = parser.to_xml(data=execute_data)
 494
 495     def stop(self):
 496        for testbed in self._testbeds.values():
 497            testbed.stop()
 498        self._unpersist_testbed_proxies()
 499
 500     def recover(self):
 501         # reload perviously persisted testbed access configurations
 502         self._load_testbed_proxies()
 503
 504         # recreate testbed proxies by reconnecting only
 505         self._init_testbed_controllers(recover = True)
 506
 507         # another time, for netrefs
 508         self._init_testbed_controllers(recover = True)
 509
 510     def is_finished(self, guid):
 511         testbed = self._testbed_for_guid(guid)
 512         if testbed != None:
 513             return testbed.status(guid) == AS.STATUS_FINISHED
 514         raise RuntimeError("No element exists with guid %d" % guid)
 515
 516     def set(self, guid, name, value, time = TIME_NOW):
 517         testbed = self._testbed_for_guid(guid)
 518         if testbed != None:
 519             testbed.set(guid, name, value, time)
 520         else:
 521             raise RuntimeError("No element exists with guid %d" % guid)
 522
 523     def get(self, guid, name, time = TIME_NOW):
 524         testbed = self._testbed_for_guid(guid)
 525         if testbed != None:
 526             return testbed.get(guid, name, time)
 527         raise RuntimeError("No element exists with guid %d" % guid)
 528
 529     def get_deferred(self, guid, name, time = TIME_NOW):
 530         testbed = self._testbed_for_guid(guid)
 531         if testbed != None:
 532             return testbed.get_deferred(guid, name, time)
 533         raise RuntimeError("No element exists with guid %d" % guid)
 534
 535     def get_factory_id(self, guid):
 536         testbed = self._testbed_for_guid(guid)
 537         if testbed != None:
 538             return testbed.get_factory_id(guid)
 539         raise RuntimeError("No element exists with guid %d" % guid)
 540
 541     def get_testbed_id(self, guid):
 542         testbed = self._testbed_for_guid(guid)
 543         if testbed != None:
 544             return testbed.testbed_id
 545         raise RuntimeError("No element exists with guid %d" % guid)
 546
 547     def get_testbed_version(self, guid):
 548         testbed = self._testbed_for_guid(guid)
 549         if testbed != None:
 550             return testbed.testbed_version
 551         raise RuntimeError("No element exists with guid %d" % guid)
 552
 553     def shutdown(self):
 554         exceptions = list()
 555         for testbed in self._testbeds.values():
 556             try:
 557                 testbed.shutdown()
 558             except:
 559                 exceptions.append(sys.exc_info())
 560         for exc_info in exceptions:
 561             raise exc_info[0], exc_info[1], exc_info[2]
 562
 563     def _testbed_for_guid(self, guid):
 564         for testbed_guid in self._testbeds.keys():
 565             if guid in self._guids_in_testbed(testbed_guid):
 566                 return self._testbeds[testbed_guid]
 567         return None
 568
 569     def _guids_in_testbed(self, testbed_guid):
 570         if testbed_guid not in self._testbeds:
 571             return set()
 572         if testbed_guid not in self._guids_in_testbed_cache:
 573             self._guids_in_testbed_cache[testbed_guid] = \
 574                 set(self._testbeds[testbed_guid].guids)
 575         return self._guids_in_testbed_cache[testbed_guid]
 576
 577     @staticmethod
 578     def _netref_component_split(component):
 579         match = COMPONENT_PATTERN.match(component)
 580         if match:
 581             return match.group("kind"), match.group("index")
 582         else:
 583             return component, None
 584
 585     _NETREF_COMPONENT_GETTERS = {
 586         'addr':
 587             lambda testbed, guid, index, name:
 588                 testbed.get_address(guid, int(index), name),
 589         'route' :
 590             lambda testbed, guid, index, name:
 591                 testbed.get_route(guid, int(index), name),
 592         'trace' :
 593             lambda testbed, guid, index, name:
 594                 testbed.trace(guid, index, name),
 595         '' :
 596             lambda testbed, guid, index, name:
 597                 testbed.get(guid, name),
 598     }
 599
 600     def resolve_netref_value(self, value, failval = None):
 601         match = ATTRIBUTE_PATTERN_BASE.search(value)
 602         if match:
 603             label = match.group("label")
 604             if label.startswith('GUID-'):
 605                 ref_guid = int(label[5:])
 606                 if ref_guid:
 607                     expr = match.group("expr")
 608                     component = (match.group("component") or "")[1:] # skip the dot
 609                     attribute = match.group("attribute")
 610
 611                     # split compound components into component kind and index
 612                     # eg: 'addr[0]' -> ('addr', '0')
 613                     component, component_index = self._netref_component_split(component)
 614
 615                     # find object and resolve expression
 616                     for ref_testbed_guid, ref_testbed in self._testbeds.iteritems():
 617                         if component not in self._NETREF_COMPONENT_GETTERS:
 618                             raise ValueError, "Malformed netref: %r - unknown component" % (expr,)
 619                         elif ref_guid not in self._guids_in_testbed(ref_testbed_guid):
 620                             pass
 621                         else:
 622                             ref_value = self._NETREF_COMPONENT_GETTERS[component](
 623                                 ref_testbed, ref_guid, component_index, attribute)
 624                             if ref_value:
 625                                 return value.replace(match.group(), ref_value)
 626         # couldn't find value
 627         return failval
 628
 629     def do_netrefs(self, data, fail_if_undefined = False):
 630         # element netrefs
 631         for (testbed_guid, guid), attrs in self._netrefs.items():
 632             testbed = self._testbeds.get(testbed_guid)
 633             if testbed is not None:
 634                 for name in set(attrs):
 635                     value = testbed.get(guid, name)
 636                     if isinstance(value, basestring):
 637                         ref_value = self.resolve_netref_value(value)
 638                         if ref_value is not None:
 639                             testbed.set(guid, name, ref_value)
 640                             attrs.remove(name)
 641                         elif fail_if_undefined:
 642                             raise ValueError, "Unresolvable netref in: %r=%r" % (name,value,)
 643                 if not attrs:
 644                     del self._netrefs[(testbed_guid, guid)]
 645
 646         # testbed netrefs
 647         for testbed_guid, attrs in self._testbed_netrefs.items():
 648             tb_data = dict(data.get_attribute_data(testbed_guid))
 649             if data:
 650                 for name in set(attrs):
 651                     value = tb_data.get(name)
 652                     if isinstance(value, basestring):
 653                         ref_value = self.resolve_netref_value(value)
 654                         if ref_value is not None:
 655                             data.set_attribute_data(testbed_guid, name, ref_value)
 656                             attrs.remove(name)
 657                         elif fail_if_undefined:
 658                             raise ValueError, "Unresolvable netref in: %r" % (value,)
 659                 if not attrs:
 660                     del self._testbed_netrefs[testbed_guid]
 661
 662
 663     def _init_testbed_controllers(self, data, recover = False):
 664         blacklist_testbeds = set(self._testbeds)
 665         element_guids = list()
 666         label_guids = dict()
 667         data_guids = data.guids
 668
 669         # create testbed controllers
 670         for guid in data_guids:
 671             if data.is_testbed_data(guid):
 672                 if guid not in self._testbeds:
 673                     self._create_testbed_controller(guid, data, element_guids,
 674                             recover)
 675             else:
 676                 (testbed_guid, factory_id) = data.get_box_data(guid)
 677                 if testbed_guid not in blacklist_testbeds:
 678                     element_guids.append(guid)
 679                     label = data.get_attribute_data(guid, "label")
 680                     if label is not None:
 681                         if label in label_guids:
 682                             raise RuntimeError, "Label %r is not unique" % (label,)
 683                         label_guids[label] = guid
 684
 685         # replace references to elements labels for its guid
 686         self._resolve_labels(data, data_guids, label_guids)
 687
 688         # program testbed controllers
 689         if not recover:
 690             self._program_testbed_controllers(element_guids, data)
 691
 692     def _resolve_labels(self, data, data_guids, label_guids):
 693         netrefs = self._netrefs
 694         testbed_netrefs = self._testbed_netrefs
 695         for guid in data_guids:
 696             for name, value in data.get_attribute_data(guid):
 697                 if isinstance(value, basestring):
 698                     match = ATTRIBUTE_PATTERN_BASE.search(value)
 699                     if match:
 700                         label = match.group("label")
 701                         if not label.startswith('GUID-'):
 702                             ref_guid = label_guids.get(label)
 703                             if ref_guid is not None:
 704                                 value = ATTRIBUTE_PATTERN_BASE.sub(
 705                                     ATTRIBUTE_PATTERN_GUID_SUB % dict(
 706                                         guid = 'GUID-%d' % (ref_guid,),
 707                                         expr = match.group("expr"),
 708                                         label = label),
 709                                     value)
 710                                 data.set_attribute_data(guid, name, value)
 711
 712                                 # memorize which guid-attribute pairs require
 713                                 # postprocessing, to avoid excessive controller-testbed
 714                                 # communication at configuration time
 715                                 # (which could require high-latency network I/O)
 716                                 if not data.is_testbed_data(guid):
 717                                     (testbed_guid, factory_id) = data.get_box_data(guid)
 718                                     netrefs[(testbed_guid, guid)].add(name)
 719                                 else:
 720                                     testbed_netrefs[guid].add(name)
 721
 722     def _create_testbed_controller(self, guid, data, element_guids, recover):
 723         (testbed_id, testbed_version) = data.get_testbed_data(guid)
 724         deployment_config = self._deployment_config.get(guid)
 725
 726         # deferred import because proxy needs
 727         # our class definitions to define proxies
 728         import nepi.util.proxy as proxy
 729
 730         if deployment_config is None:
 731             # need to create one
 732             deployment_config = proxy.AccessConfiguration()
 733
 734             for (name, value) in data.get_attribute_data(guid):
 735                 if value is not None and deployment_config.has_attribute(name):
 736                     # if any deployment config attribute has a netref, we can't
 737                     # create this controller yet
 738                     if isinstance(value, basestring) and ATTRIBUTE_PATTERN_BASE.search(value):
 739                         # remember to re-issue this one
 740                         self._netreffed_testbeds.add(guid)
 741                         return
 742
 743                     # copy deployment config attribute
 744                     deployment_config.set_attribute_value(name, value)
 745
 746             # commit config
 747             self._deployment_config[guid] = deployment_config
 748
 749         if deployment_config is not None:
 750             # force recovery mode
 751             deployment_config.set_attribute_value("recover",recover)
 752
 753         testbed = proxy.create_testbed_controller(testbed_id, testbed_version,
 754                 deployment_config)
 755         for (name, value) in data.get_attribute_data(guid):
 756             testbed.defer_configure(name, value)
 757         self._testbeds[guid] = testbed
 758         if guid in self._netreffed_testbeds:
 759             self._netreffed_testbeds.remove(guid)
 760
 761     def _program_testbed_controllers(self, element_guids, data):
 762         deferred = dict()
 763
 764         def add_deferred_testbed(deferred, testbed_guid):
 765             if not testbed_guid in deferred:
 766                 deferred[testbed_guid] = dict()
 767                 deferred[testbed_guid]["connections"] = set()
 768                 deferred[testbed_guid]["cross_connections"] = set()
 769
 770         def add_deferred_connection(deferred, data, guid, connector_type_name,
 771                 other_guid, other_connector_type_name):
 772
 773             (testbed_guid, factory_id) = data.get_box_data(guid)
 774             (other_testbed_guid, other_factory_id) = data.get_box_data(
 775                     other_guid)
 776             testbed = self._testbeds[testbed_guid]
 777             testbed_id = testbed.testbed_id
 778             add_deferred_testbed(deferred, testbed_guid)
 779
 780             if testbed_guid == other_testbed_guid:
 781                 # each testbed should take care of enforcing internal
 782                 # connection simmetry, so each connection is only
 783                 # added in one direction
 784                 c = (guid, connector_type_name, other_guid,
 785                         other_connector_type_name)
 786                 deferred[testbed_guid]["connections"].add(c)
 787             else:
 788                 # the controller takes care of cross_connection simmetry
 789                 # so cross_connections are added in both directions
 790                 other_testbed = self._testbeds[other_testbed_guid]
 791                 other_testbed_id = other_testbed.testbed_id
 792                 add_deferred_testbed(deferred, other_testbed_guid)
 793                 c1 = (testbed_guid, testbed_id, guid, factory_id,
 794                         connector_type_name, other_testbed_guid,
 795                         other_testbed_id, other_guid, other_factory_id,
 796                         other_connector_type_name)
 797                 c2 = (other_testbed_guid, other_testbed_id, other_guid,
 798                         other_factory_id, other_connector_type_name,
 799                         testbed_guid, testbed_id, guid, factory_id,
 800                         connector_type_name)
 801                 deferred[testbed_guid]["cross_connections"].add(c1)
 802                 deferred[other_testbed_guid]["cross_connections"].add(c2)
 803
 804         def resolve_create_netref(data, guid, name, value):
 805             # Try to resolve create-time netrefs, if possible
 806             if isinstance(value, basestring) and ATTRIBUTE_PATTERN_BASE.search(value):
 807                 try:
 808                     nuvalue = self.resolve_netref_value(value)
 809                 except:
 810                     # Any trouble means we're not in shape to resolve the netref yet
 811                     nuvalue = None
 812                 if nuvalue is not None:
 813                     # Only if we succeed we remove the netref deferral entry
 814                     value = nuvalue
 815                     data.set_attribute_data(guid, name, value)
 816                     if (testbed_guid, guid) in self._netrefs:
 817                         self._netrefs[(testbed_guid, guid)].discard(name)
 818             return value
 819
 820         for guid in element_guids:
 821             (testbed_guid, factory_id) = data.get_box_data(guid)
 822             testbed = self._testbeds.get(testbed_guid)
 823             # create
 824             testbed.defer_create(guid, factory_id)
 825             # set attributes
 826             for (name, value) in data.get_attribute_data(guid):
 827                 value = resolve_create_netref(data, guid, name, value)
 828                 testbed.defer_create_set(guid, name, value)
 829             # traces
 830             for trace_id in data.get_trace_data(guid):
 831                 testbed.defer_add_trace(guid, trace_id)
 832             # addresses
 833             for (address, netprefix, broadcast) in data.get_address_data(guid):
 834                 if address != None:
 835                     testbed.defer_add_address(guid, address, netprefix,
 836                             broadcast)
 837             # routes
 838             for (destination, netprefix, nexthop, metric) in data.get_route_data(guid):
 839                 testbed.defer_add_route(guid, destination, netprefix, nexthop, metric)
 840             # store connections data
 841             for (connector_type_name, other_guid, other_connector_type_name) \
 842                     in data.get_connection_data(guid):
 843                 add_deferred_connection(deferred, data, guid,
 844                         connector_type_name, other_guid,
 845                         other_connector_type_name)
 846
 847         # connections
 848         for testbed_guid, data in deferred.iteritems():
 849             testbed = self._testbeds.get(testbed_guid)
 850             for (guid, connector_type_name, other_guid,
 851                     other_connector_type_name) in data["connections"]:
 852                 testbed.defer_connect(guid, connector_type_name,
 853                         other_guid, other_connector_type_name)
 854             for (testbed_guid, testbed_id, guid, factory_id,
 855                         connector_type_name, other_testbed_guid,
 856                         other_testbed_id, other_guid, other_factory_id,
 857                         other_connector_type_name) in data["cross_connections"]:
 858                 testbed.defer_cross_connect(guid, connector_type_name, other_guid,
 859                         other_testbed_guid, other_testbed_id, other_factory_id,
 860                         other_connector_type_name)
 861                 # save cross data for later
 862                 self._add_crossdata(testbed_guid, guid, other_testbed_guid,
 863                         other_guid)
 864
 865     def _add_crossdata(self, testbed_guid, guid, cross_testbed_guid, cross_guid):
 866         if testbed_guid not in self._cross_data:
 867             self._cross_data[testbed_guid] = dict()
 868         if cross_testbed_guid not in self._cross_data[testbed_guid]:
 869             self._cross_data[testbed_guid][cross_testbed_guid] = set()
 870         self._cross_data[testbed_guid][cross_testbed_guid].add(cross_guid)
 871
 872     def _get_cross_data(self, testbed_guid):
 873         cross_data = dict()
 874         if not testbed_guid in self._cross_data:
 875             return cross_data
 876         for cross_testbed_guid, guid_list in \
 877                 self._cross_data[testbed_guid].iteritems():
 878             cross_data[cross_testbed_guid] = dict()
 879             cross_testbed = self._testbeds[cross_testbed_guid]
 880             for cross_guid in guid_list:
 881                 elem_cross_data = dict(
 882                     _guid = cross_guid,
 883                     _testbed_guid = cross_testbed_guid,
 884                     _testbed_id = cross_testbed.testbed_id,
 885                     _testbed_version = cross_testbed.testbed_version)
 886                 cross_data[cross_testbed_guid][cross_guid] = elem_cross_data
 887                 attribute_list = cross_testbed.get_attribute_list(cross_guid)
 888                 for attr_name in attribute_list:
 889                     attr_value = cross_testbed.get_deferred(cross_guid, attr_name)
 890                     elem_cross_data[attr_name] = attr_value
 891         return cross_data
 892