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