testbed_impl.get improved
[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 testbed_id(self):
169         return self._testbed_id
170
171     @property
172     def testbed_version(self):
173         return self._testbed_version
174
175     @property
176     def guids(self):
177         raise NotImplementedError
178
179     def defer_configure(self, name, value):
180         """Instructs setting a configuartion attribute for the testbed instance"""
181         raise NotImplementedError
182
183     def defer_create(self, guid, factory_id):
184         """Instructs creation of element """
185         raise NotImplementedError
186
187     def defer_create_set(self, guid, name, value):
188         """Instructs setting an initial attribute on an element"""
189         raise NotImplementedError
190
191     def defer_factory_set(self, guid, name, value):
192         """Instructs setting an attribute on a factory"""
193         raise NotImplementedError
194
195     def defer_connect(self, guid1, connector_type_name1, guid2, 
196             connector_type_name2): 
197         """Instructs creation of a connection between the given connectors"""
198         raise NotImplementedError
199
200     def defer_cross_connect(self, guid, connector_type_name, cross_guid, 
201             cross_testbed_id, cross_factory_id, cross_connector_type_name):
202         """
203         Instructs creation of a connection between the given connectors 
204         of different testbed instances
205         """
206         raise NotImplementedError
207
208     def defer_add_trace(self, guid, trace_id):
209         """Instructs the addition of a trace"""
210         raise NotImplementedError
211
212     def defer_add_address(self, guid, address, netprefix, broadcast): 
213         """Instructs the addition of an address"""
214         raise NotImplementedError
215
216     def defer_add_route(self, guid, destination, netprefix, nexthop):
217         """Instructs the addition of a route"""
218         raise NotImplementedError
219
220     def do_setup(self):
221         """After do_setup the testbed initial configuration is done"""
222         raise NotImplementedError
223
224     def do_create(self):
225         """
226         After do_create all instructed elements are created and 
227         attributes setted
228         """
229         raise NotImplementedError
230
231     def do_connect_init(self):
232         """
233         After do_connect_init all internal connections between testbed elements
234         are initiated
235         """
236         raise NotImplementedError
237
238     def do_connect_compl(self):
239         """
240         After do_connect all internal connections between testbed elements
241         are completed
242         """
243         raise NotImplementedError
244
245     def do_configure(self):
246         """After do_configure elements are configured"""
247         raise NotImplementedError
248
249     def do_cross_connect_init(self, cross_data):
250         """
251         After do_cross_connect_init initiation of all external connections 
252         between different testbed elements is performed
253         """
254         raise NotImplementedError
255
256     def do_cross_connect_compl(self, cross_data):
257         """
258         After do_cross_connect_compl completion of all external connections 
259         between different testbed elements is performed
260         """
261         raise NotImplementedError
262
263     def start(self):
264         raise NotImplementedError
265
266     def stop(self):
267         raise NotImplementedError
268
269     def set(self, time, guid, name, value):
270         raise NotImplementedError
271
272     def get(self, time, guid, name):
273         raise NotImplementedError
274     
275     def get_route(self, guid, index, attribute):
276         """
277         Params:
278             
279             guid: guid of box to query
280             index: number of routing entry to fetch
281             attribute: one of Destination, NextHop, NetPrefix
282         """
283         raise NotImplementedError
284
285     def get_address(self, guid, index, attribute='Address'):
286         """
287         Params:
288             
289             guid: guid of box to query
290             index: number of inteface to select
291             attribute: one of Address, NetPrefix, Broadcast
292         """
293         raise NotImplementedError
294
295     def get_attribute_list(self, guid):
296         raise NotImplementedError
297
298     def action(self, time, guid, action):
299         raise NotImplementedError
300
301     def status(self, guid):
302         raise NotImplementedError
303
304     def trace(self, guid, trace_id, attribute='value'):
305         raise NotImplementedError
306
307     def shutdown(self):
308         raise NotImplementedError
309
310 class ExperimentController(object):
311     def __init__(self, experiment_xml, root_dir):
312         self._experiment_xml = experiment_xml
313         self._testbeds = dict()
314         self._access_config = dict()
315         self._netrefs = dict()
316         self._cross_data = dict()
317         self._root_dir = root_dir
318
319         self.persist_experiment_xml()
320
321     @property
322     def experiment_xml(self):
323         return self._experiment_xml
324
325     def persist_experiment_xml(self):
326         xml_path = os.path.join(self._root_dir, "experiment.xml")
327         f = open(xml_path, "w")
328         f.write(self._experiment_xml)
329         f.close()
330
331     def set_access_configuration(self, testbed_guid, access_config):
332         self._access_config[testbed_guid] = access_config
333
334     def trace(self, testbed_guid, guid, trace_id, attribute='value'):
335         return self._testbeds[testbed_guid].trace(guid, trace_id, attribute)
336
337     @staticmethod
338     def _parallel(callables):
339         threads = [ threading.Thread(target=callable) for callable in callables ]
340         for thread in threads:
341             thread.start()
342         for thread in threads:
343             thread.join()
344
345     def start(self):
346         self._init_testbed_controllers()
347         
348         # persist testbed connection data, for potential recovery
349         self._persist_testbed_proxies()
350         
351         # perform setup in parallel for all test beds,
352         # wait for all threads to finish
353         self._parallel([testbed.do_setup 
354                         for testbed in self._testbeds.itervalues()])
355         
356         # perform create-connect in parallel, wait
357         # (internal connections only)
358         self._parallel([testbed.do_create
359                         for testbed in self._testbeds.itervalues()])
360
361         self._parallel([testbed.do_connect_init
362                         for testbed in self._testbeds.itervalues()])
363
364         self._parallel([testbed.do_connect_compl
365                         for testbed in self._testbeds.itervalues()])
366
367         self._parallel([testbed.do_preconfigure
368                         for testbed in self._testbeds.itervalues()])
369
370         # resolve netrefs
371         self.do_netrefs(fail_if_undefined=True)
372         
373         # perform do_configure in parallel for al testbeds
374         # (it's internal configuration for each)
375         self._parallel([testbed.do_configure
376                         for testbed in self._testbeds.itervalues()])
377
378         # cross-connect (cannot be done in parallel)
379         for guid, testbed in self._testbeds.iteritems():
380             cross_data = self._get_cross_data(guid)
381             testbed.do_cross_connect_init(cross_data)
382         for guid, testbed in self._testbeds.iteritems():
383             cross_data = self._get_cross_data(guid)
384             testbed.do_cross_connect_compl(cross_data)
385        
386         # start experiment (parallel start on all testbeds)
387         self._parallel([testbed.start
388                         for testbed in self._testbeds.itervalues()])
389
390     def _persist_testbed_proxies(self):
391         TRANSIENT = ('Recover',)
392         
393         # persist access configuration for all testbeds, so that
394         # recovery mode can reconnect to them if it becomes necessary
395         conf = ConfigParser.RawConfigParser()
396         for testbed_guid, testbed_config in self._access_config.iteritems():
397             testbed_guid = str(testbed_guid)
398             conf.add_section(testbed_guid)
399             for attr in testbed_config.attributes_list:
400                 if attr not in TRANSIENT:
401                     conf.set(testbed_guid, attr, 
402                         testbed_config.get_attribute_value(attr))
403         
404         f = open(os.path.join(self._root_dir, 'access_config.ini'), 'w')
405         conf.write(f)
406         f.close()
407     
408     def _load_testbed_proxies(self):
409         TYPEMAP = {
410             STRING : 'get',
411             INTEGER : 'getint',
412             FLOAT : 'getfloat',
413             BOOLEAN : 'getboolean',
414         }
415         
416         conf = ConfigParser.RawConfigParser()
417         conf.read(os.path.join(self._root_dir, 'access_config.ini'))
418         for testbed_guid in conf.sections():
419             testbed_config = proxy.AccessConfiguration()
420             for attr in conf.options(testbed_guid):
421                 testbed_config.set_attribute_value(attr, 
422                     conf.get(testbed_guid, attr) )
423                 
424             testbed_guid = str(testbed_guid)
425             conf.add_section(testbed_guid)
426             for attr in testbed_config.attributes_list:
427                 if attr not in TRANSIENT:
428                     getter = getattr(conf, TYPEMAP.get(
429                         testbed_config.get_attribute_type(attr),
430                         'get') )
431                     testbed_config.set_attribute_value(
432                         testbed_guid, attr, getter(attr))
433     
434     def _unpersist_testbed_proxies(self):
435         try:
436             os.remove(os.path.join(self._root_dir, 'access_config.ini'))
437         except:
438             # Just print exceptions, this is just cleanup
439             import traceback
440             traceback.print_exc(file=sys.stderr)
441
442     def stop(self):
443        for testbed in self._testbeds.values():
444            testbed.stop()
445        self._unpersist_testbed_proxies()
446    
447     def recover(self):
448         # reload perviously persisted testbed access configurations
449         self._load_testbed_proxies()
450         
451         # recreate testbed proxies by reconnecting only
452         self._init_testbed_controllers(recover = True)
453
454     def is_finished(self, guid):
455         for testbed in self._testbeds.values():
456             for guid_ in testbed.guids:
457                 if guid_ == guid:
458                     return testbed.status(guid) == STATUS_FINISHED
459         raise RuntimeError("No element exists with guid %d" % guid)    
460
461     def shutdown(self):
462        for testbed in self._testbeds.values():
463            testbed.shutdown()
464
465     @staticmethod
466     def _netref_component_split(component):
467         match = COMPONENT_PATTERN.match(component)
468         if match:
469             return match.group("kind"), match.group("index")
470         else:
471             return component, None
472
473     def do_netrefs(self, fail_if_undefined = False):
474         COMPONENT_GETTERS = {
475             'addr':
476                 lambda testbed, guid, index, name: 
477                     testbed.get_address(guid, index, name),
478             'route' :
479                 lambda testbed, guid, index, name: 
480                     testbed.get_route(guid, index, name),
481             'trace' :
482                 lambda testbed, guid, index, name: 
483                     testbed.trace(guid, index, name),
484             '' : 
485                 lambda testbed, guid, index, name: 
486                     testbed.get(TIME_NOW, guid, name),
487         }
488         
489         for (testbed_guid, guid), attrs in self._netrefs.iteritems():
490             testbed = self._testbeds[testbed_guid]
491             for name in attrs:
492                 value = testbed.get(TIME_NOW, guid, name)
493                 if isinstance(value, basestring):
494                     match = ATTRIBUTE_PATTERN_BASE.search(value)
495                     if match:
496                         label = match.group("label")
497                         if label.startswith('GUID-'):
498                             ref_guid = int(label[5:])
499                             if ref_guid:
500                                 expr = match.group("expr")
501                                 component = match.group("component")[1:] # skip the dot
502                                 attribute = match.group("attribute")
503                                 
504                                 # split compound components into component kind and index
505                                 # eg: 'addr[0]' -> ('addr', '0')
506                                 component, component_index = self._netref_component_split(component)
507                                 
508                                 # find object and resolve expression
509                                 for ref_testbed in self._testbeds.itervalues():
510                                     if component not in COMPONENT_GETTERS:
511                                         raise ValueError, "Malformed netref: %r - unknown component" % (expr,)
512                                     else:
513                                         ref_value = COMPONENT_GETTERS[component](
514                                             ref_testbed, ref_guid, component_index, attribute)
515                                         if ref_value:
516                                             testbed.set(TIME_NOW, guid, name, 
517                                                 value.replace(match.group(), ref_value))
518                                             break
519                                 else:
520                                     # couldn't find value
521                                     if fail_if_undefined:
522                                         raise ValueError, "Unresolvable GUID: %r, in netref: %r" % (ref_guid, expr)
523
524     def _init_testbed_controllers(self, recover = False):
525         parser = XmlExperimentParser()
526         data = parser.from_xml_to_data(self._experiment_xml)
527         element_guids = list()
528         label_guids = dict()
529         data_guids = data.guids
530
531         # create testbed controllers
532         for guid in data_guids:
533             if data.is_testbed_data(guid):
534                 self._create_testbed_controller(guid, data, element_guids, 
535                         recover)
536             else:
537                 element_guids.append(guid)
538                 label = data.get_attribute_data(guid, "label")
539                 if label is not None:
540                     if label in label_guids:
541                         raise RuntimeError, "Label %r is not unique" % (label,)
542                     label_guids[label] = guid
543
544         # replace references to elements labels for its guid
545         self._resolve_labels(data, data_guids, label_guids)
546     
547         # program testbed controllers
548         if not recover:
549             self._program_testbed_controllers(element_guids, data)
550
551     def _resolve_labels(self, data, data_guids, label_guids):
552         netrefs = self._netrefs
553         for guid in data_guids:
554             if not data.is_testbed_data(guid):
555                 for name, value in data.get_attribute_data(guid):
556                     if isinstance(value, basestring):
557                         match = ATTRIBUTE_PATTERN_BASE.search(value)
558                         if match:
559                             label = match.group("label")
560                             if not label.startswith('GUID-'):
561                                 ref_guid = label_guids.get(label)
562                                 if ref_guid is not None:
563                                     value = ATTRIBUTE_PATTERN_BASE.sub(
564                                         ATTRIBUTE_PATTERN_GUID_SUB % dict(
565                                             guid = 'GUID-%d' % (ref_guid,),
566                                             expr = match.group("expr"),
567                                             label = label), 
568                                         value)
569                                     data.set_attribute_data(guid, name, value)
570                                     
571                                     # memorize which guid-attribute pairs require
572                                     # postprocessing, to avoid excessive controller-testbed
573                                     # communication at configuration time
574                                     # (which could require high-latency network I/O)
575                                     (testbed_guid, factory_id) = data.get_box_data(guid)
576                                     netrefs.setdefault((testbed_guid, guid), set()).add(name)
577
578     def _create_testbed_controller(self, guid, data, element_guids, recover):
579         (testbed_id, testbed_version) = data.get_testbed_data(guid)
580         access_config = None if guid not in self._access_config else\
581                 self._access_config[guid]
582         
583         if recover and access_config is None:
584             # need to create one
585             access_config = self._access_config[guid] = proxy.AccessConfiguration()
586         if access_config is not None:
587             # force recovery mode 
588             access_config.set_attribute_value("recover",recover)
589         
590         testbed = proxy.create_testbed_controller(testbed_id, 
591                 testbed_version, access_config)
592         for (name, value) in data.get_attribute_data(guid):
593             testbed.defer_configure(name, value)
594         self._testbeds[guid] = testbed
595
596     def _program_testbed_controllers(self, element_guids, data):
597         for guid in element_guids:
598             (testbed_guid, factory_id) = data.get_box_data(guid)
599             testbed = self._testbeds[testbed_guid]
600             testbed.defer_create(guid, factory_id)
601             for (name, value) in data.get_attribute_data(guid):
602                 testbed.defer_create_set(guid, name, value)
603
604         for guid in element_guids: 
605             (testbed_guid, factory_id) = data.get_box_data(guid)
606             testbed = self._testbeds[testbed_guid]
607             for (connector_type_name, cross_guid, cross_connector_type_name) \
608                     in data.get_connection_data(guid):
609                 (testbed_guid, factory_id) = data.get_box_data(guid)
610                 (cross_testbed_guid, cross_factory_id) = data.get_box_data(
611                         cross_guid)
612                 if testbed_guid == cross_testbed_guid:
613                     testbed.defer_connect(guid, connector_type_name, 
614                             cross_guid, cross_connector_type_name)
615                 else: 
616                     cross_testbed = self._testbeds[cross_testbed_guid]
617                     cross_testbed_id = cross_testbed.testbed_id
618                     testbed.defer_cross_connect(guid, connector_type_name, cross_guid, 
619                             cross_testbed_id, cross_factory_id, 
620                             cross_connector_type_name)
621                     # save cross data for later
622                     self._add_crossdata(testbed_guid, guid, cross_testbed_guid,
623                             cross_guid)
624             for trace_id in data.get_trace_data(guid):
625                 testbed.defer_add_trace(guid, trace_id)
626             for (autoconf, address, netprefix, broadcast) in \
627                     data.get_address_data(guid):
628                 if address != None:
629                     testbed.defer_add_address(guid, address, netprefix, 
630                             broadcast)
631             for (destination, netprefix, nexthop) in data.get_route_data(guid):
632                 testbed.defer_add_route(guid, destination, netprefix, nexthop)
633                 
634     def _add_crossdata(self, testbed_guid, guid, cross_testbed_guid, cross_guid):
635         if testbed_guid not in self._cross_data:
636             self._cross_data[testbed_guid] = dict()
637         if cross_testbed_guid not in self._cross_data[testbed_guid]:
638             self._cross_data[testbed_guid][cross_testbed_guid] = list()
639         if cross_testbed_guid not in self._cross_data:
640             self._cross_data[cross_testbed_guid] = dict()
641         if testbed_guid not in self._cross_data[cross_testbed_guid]:
642             self._cross_data[cross_testbed_guid][testbed_guid] = list()
643         self._cross_data[testbed_guid][cross_testbed_guid].append(cross_guid)
644         self._cross_data[cross_testbed_guid][testbed_guid].append(guid)
645
646     def _get_cross_data(self, testbed_guid):
647         cross_data = dict()
648         if not testbed_guid in self._cross_data:
649             return cross_data
650         for cross_testbed_guid, guid_list in \
651                 self._cross_data[testbed_guid].iteritems():
652             cross_data[cross_testbed_guid] = dict()
653             cross_testbed = self._testbeds[cross_testbed_guid]
654             for cross_guid in guid_list:
655                 elem_cross_data = dict()
656                 cross_data[cross_testbed_guid][cross_guid] = elem_cross_data
657                 attributes_list = cross_testbed.get_attribute_list(cross_guid)
658                 for attr_name in attributes_list:
659                     attr_value = cross_testbed.get(TIME_NOW, cross_guid, 
660                             attr_name)
661                     elem_cross_data[attr_name] = attr_value
662         return elem_cross_data
663