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