X-Git-Url: https://gerrit.fd.io/r/gitweb?a=blobdiff_plain;f=src%2Fvpp-api%2Fpython%2Fvpp_papi%2Fvpp_papi.py;h=9c4ede90d48d4ae056f8d6f79e397b23f88af42a;hb=8006c6a;hp=6b6b79fd70b9cb0d721750f5be74a01f24b457ce;hpb=7cd468a3d7dee7d6c92f69a0bb7061ae208ec727;p=vpp.git diff --git a/src/vpp-api/python/vpp_papi/vpp_papi.py b/src/vpp-api/python/vpp_papi/vpp_papi.py index 6b6b79fd70b..9c4ede90d48 100644 --- a/src/vpp-api/python/vpp_papi/vpp_papi.py +++ b/src/vpp-api/python/vpp_papi/vpp_papi.py @@ -15,436 +15,692 @@ # from __future__ import print_function -import sys, os, logging, collections, struct, json, threading, glob -logging.basicConfig(level=logging.DEBUG) -import vpp_api +from __future__ import absolute_import +import sys +import os +import logging +import collections +import struct +import json +import threading +import fnmatch +import weakref +import atexit +from . vpp_serializer import VPPType, VPPEnumType, VPPUnionType, BaseTypes +from . vpp_serializer import VPPMessage, vpp_get_type, VPPTypeAlias +from . macaddress import MACAddress, mac_pton, mac_ntop -def eprint(*args, **kwargs): - print(*args, file=sys.stderr, **kwargs) +logger = logging.getLogger(__name__) + +if sys.version[0] == '2': + import Queue as queue +else: + import queue as queue + + +class VppEnumType(type): + def __getattr__(cls, name): + t = vpp_get_type(name) + return t.enum + + +# Python3 +# class VppEnum(metaclass=VppEnumType): +# pass +class VppEnum(object): + __metaclass__ = VppEnumType + + +def vpp_atexit(vpp_weakref): + """Clean up VPP connection on shutdown.""" + vpp_instance = vpp_weakref() + if vpp_instance and vpp_instance.transport.connected: + vpp_instance.logger.debug('Cleaning up VPP on exit') + vpp_instance.disconnect() + + +if sys.version[0] == '2': + def vpp_iterator(d): + return d.iteritems() +else: + def vpp_iterator(d): + return d.items() + + +def call_logger(msgdef, kwargs): + s = 'Calling {}('.format(msgdef.name) + for k, v in kwargs.items(): + s += '{}:{} '.format(k, v) + s += ')' + return s + + +def return_logger(r): + s = 'Return from {}'.format(r) + return s + + +class VppApiDynamicMethodHolder(object): + pass + + +class FuncWrapper(object): + def __init__(self, func): + self._func = func + self.__name__ = func.__name__ + + def __call__(self, **kwargs): + return self._func(**kwargs) + + +class VPPApiError(Exception): + pass + + +class VPPNotImplementedError(NotImplementedError): + pass + + +class VPPIOError(IOError): + pass + + +class VPPRuntimeError(RuntimeError): + pass + + +class VPPValueError(ValueError): + pass + + +class VPP(object): + """VPP interface. + + This class provides the APIs to VPP. The APIs are loaded + from provided .api.json files and makes functions accordingly. + These functions are documented in the VPP .api files, as they + are dynamically created. + + Additionally, VPP can send callback messages; this class + provides a means to register a callback function to receive + these messages in a background thread. + """ + VPPApiError = VPPApiError + VPPRuntimeError = VPPRuntimeError + VPPValueError = VPPValueError + VPPNotImplementedError = VPPNotImplementedError + VPPIOError = VPPIOError + + def process_json_file(self, apidef_file): + api = json.load(apidef_file) + types = {} + for t in api['enums']: + t[0] = 'vl_api_' + t[0] + '_t' + types[t[0]] = {'type': 'enum', 'data': t} + for t in api['unions']: + t[0] = 'vl_api_' + t[0] + '_t' + types[t[0]] = {'type': 'union', 'data': t} + for t in api['types']: + t[0] = 'vl_api_' + t[0] + '_t' + types[t[0]] = {'type': 'type', 'data': t} + for t, v in api['aliases'].items(): + types['vl_api_' + t + '_t'] = {'type': 'alias', 'data': v} + self.services.update(api['services']) + + i = 0 + while True: + unresolved = {} + for k, v in types.items(): + t = v['data'] + if not vpp_get_type(k): + if v['type'] == 'enum': + try: + VPPEnumType(t[0], t[1:]) + except ValueError: + unresolved[k] = v + elif v['type'] == 'union': + try: + VPPUnionType(t[0], t[1:]) + except ValueError: + unresolved[k] = v + elif v['type'] == 'type': + try: + VPPType(t[0], t[1:]) + except ValueError: + unresolved[k] = v + elif v['type'] == 'alias': + try: + VPPTypeAlias(k, t) + except ValueError: + unresolved[k] = v + if len(unresolved) == 0: + break + if i > 3: + raise VPPValueError('Unresolved type definitions {}' + .format(unresolved)) + types = unresolved + i += 1 + + for m in api['messages']: + try: + self.messages[m[0]] = VPPMessage(m[0], m[1:]) + except VPPNotImplementedError: + self.logger.error('Not implemented error for {}'.format(m[0])) + + def __init__(self, apifiles=None, testmode=False, async_thread=True, + logger=None, loglevel=None, + read_timeout=5, use_socket=False, + server_address='/run/vpp-api.sock'): + """Create a VPP API object. + + apifiles is a list of files containing API + descriptions that will be loaded - methods will be + dynamically created reflecting these APIs. If not + provided this will load the API files from VPP's + default install location. + + logger, if supplied, is the logging logger object to log to. + loglevel, if supplied, is the log level this logger is set + to report at (from the loglevels in the logging module). + """ + if logger is None: + logger = logging.getLogger(__name__) + if loglevel is not None: + logger.setLevel(loglevel) + self.logger = logger -class VPP(): - def __init__(self, apifiles = None, testmode = False): self.messages = {} + self.services = {} self.id_names = [] self.id_msgdef = [] - self.buffersize = 10000 - self.connected = False - self.header = struct.Struct('>HI') - self.results = {} - self.timeout = 5 - self.apifile = [] + self.header = VPPType('header', [['u16', 'msgid'], + ['u32', 'client_index']]) + self.apifiles = [] + self.event_callback = None + self.message_queue = queue.Queue() + self.read_timeout = read_timeout + self.async_thread = async_thread + + if use_socket: + from . vpp_transport_socket import VppTransport + else: + from . vpp_transport_shmem import VppTransport if not apifiles: # Pick up API definitions from default directory - apifiles = glob.glob('/usr/share/vpp/api/*.api.json') + try: + apifiles = self.find_api_files() + except RuntimeError: + # In test mode we don't care that we can't find the API files + if testmode: + apifiles = [] + else: + raise VPPRuntimeError for file in apifiles: - self.apifile.append(file) with open(file) as apidef_file: - api = json.load(apidef_file) - for t in api['types']: - self.add_type(t[0], t[1:]) + self.process_json_file(apidef_file) - for m in api['messages']: - self.add_message(m[0], m[1:]) + self.apifiles = apifiles # Basic sanity check if len(self.messages) == 0 and not testmode: - raise ValueError(1, 'Missing JSON message definitions') + raise VPPValueError(1, 'Missing JSON message definitions') + self.transport = VppTransport(self, read_timeout=read_timeout, + server_address=server_address) + # Make sure we allow VPP to clean up the message rings. + atexit.register(vpp_atexit, weakref.ref(self)) class ContextId(object): + """Thread-safe provider of unique context IDs.""" def __init__(self): self.context = 0 + self.lock = threading.Lock() + def __call__(self): - self.context += 1 - return self.context + """Get a new unique (or, at least, not recently used) context.""" + with self.lock: + self.context += 1 + return self.context get_context = ContextId() - def status(self): - print('Connected') if self.connected else print('Not Connected') - print('Read API definitions from', self.apifile) - - def __struct (self, t, n = None, e = -1, vl = None): - base_types = { 'u8' : 'B', - 'u16' : 'H', - 'u32' : 'I', - 'i32' : 'i', - 'u64' : 'Q', - 'f64' : 'd', - } - pack = None - if t in base_types: - pack = base_types[t] - if not vl: - if e > 0 and t == 'u8': - # Fixed byte array - return struct.Struct('>' + str(e) + 's') - if e > 0: - # Fixed array of base type - return [e, struct.Struct('>' + base_types[t])] - elif e == 0: - # Old style variable array - return [-1, struct.Struct('>' + base_types[t])] - else: - # Variable length array - return [vl, struct.Struct('>s')] if t == 'u8' else \ - [vl, struct.Struct('>' + base_types[t])] - - return struct.Struct('>' + base_types[t]) - - if t in self.messages: - ### Return a list in case of array ### - if e > 0 and not vl: - return [e, lambda self, encode, buf, offset, args: ( - self.__struct_type(encode, self.messages[t], buf, offset, - args))] - if vl: - return [vl, lambda self, encode, buf, offset, args: ( - self.__struct_type(encode, self.messages[t], buf, offset, - args))] - elif e == 0: - # Old style VLA - raise NotImplementedError(1, 'No support for compound types ' + t) - return lambda self, encode, buf, offset, args: ( - self.__struct_type(encode, self.messages[t], buf, offset, args) - ) - - raise ValueError(1, 'Invalid message type: ' + t) - - def __struct_type(self, encode, msgdef, buf, offset, kwargs): - if encode: - return self.__struct_type_encode(msgdef, buf, offset, kwargs) + def get_type(self, name): + return vpp_get_type(name) + + @classmethod + def find_api_dir(cls): + """Attempt to find the best directory in which API definition + files may reside. If the value VPP_API_DIR exists in the environment + then it is first on the search list. If we're inside a recognized + location in a VPP source tree (src/scripts and src/vpp-api/python) + then entries from there to the likely locations in build-root are + added. Finally the location used by system packages is added. + + :returns: A single directory name, or None if no such directory + could be found. + """ + dirs = [] + + if 'VPP_API_DIR' in os.environ: + dirs.append(os.environ['VPP_API_DIR']) + + # perhaps we're in the 'src/scripts' or 'src/vpp-api/python' dir; + # in which case, plot a course to likely places in the src tree + import __main__ as main + if hasattr(main, '__file__'): + # get the path of the calling script + localdir = os.path.dirname(os.path.realpath(main.__file__)) else: - return self.__struct_type_decode(msgdef, buf, offset) - - def __struct_type_encode(self, msgdef, buf, offset, kwargs): - off = offset - size = 0 - - for k in kwargs: - if k not in msgdef['args']: - raise ValueError(1, 'Invalid field-name in message call ' + k) - - for k,v in msgdef['args'].iteritems(): - off += size - if k in kwargs: - if type(v) is list: - if callable(v[1]): - e = kwargs[v[0]] if v[0] in kwargs else v[0] - size = 0 - for i in range(e): - size += v[1](self, True, buf, off + size, - kwargs[k][i]) - else: - if v[0] in kwargs: - l = kwargs[v[0]] - else: - l = len(kwargs[k]) - if v[1].size == 1: - buf[off:off + l] = bytearray(kwargs[k]) - size = l - else: - size = 0 - for i in kwargs[k]: - v[1].pack_into(buf, off + size, i) - size += v[1].size - else: - if callable(v): - size = v(self, True, buf, off, kwargs[k]) - else: - v.pack_into(buf, off, kwargs[k]) - size = v.size - else: - size = v.size if not type(v) is list else 0 - - return off + size - offset + # use cwd if there is no calling script + localdir = os.getcwd() + localdir_s = localdir.split(os.path.sep) + + def dmatch(dir): + """Match dir against right-hand components of the script dir""" + d = dir.split('/') # param 'dir' assumes a / separator + length = len(d) + return len(localdir_s) > length and localdir_s[-length:] == d + + def sdir(srcdir, variant): + """Build a path from srcdir to the staged API files of + 'variant' (typically '' or '_debug')""" + # Since 'core' and 'plugin' files are staged + # in separate directories, we target the parent dir. + return os.path.sep.join(( + srcdir, + 'build-root', + 'install-vpp%s-native' % variant, + 'vpp', + 'share', + 'vpp', + 'api', + )) + + srcdir = None + if dmatch('src/scripts'): + srcdir = os.path.sep.join(localdir_s[:-2]) + elif dmatch('src/vpp-api/python'): + srcdir = os.path.sep.join(localdir_s[:-3]) + elif dmatch('test'): + # we're apparently running tests + srcdir = os.path.sep.join(localdir_s[:-1]) + + if srcdir: + # we're in the source tree, try both the debug and release + # variants. + dirs.append(sdir(srcdir, '_debug')) + dirs.append(sdir(srcdir, '')) + + # Test for staged copies of the scripts + # For these, since we explicitly know if we're running a debug versus + # release variant, target only the relevant directory + if dmatch('build-root/install-vpp_debug-native/vpp/bin'): + srcdir = os.path.sep.join(localdir_s[:-4]) + dirs.append(sdir(srcdir, '_debug')) + if dmatch('build-root/install-vpp-native/vpp/bin'): + srcdir = os.path.sep.join(localdir_s[:-4]) + dirs.append(sdir(srcdir, '')) + + # finally, try the location system packages typically install into + dirs.append(os.path.sep.join(('', 'usr', 'share', 'vpp', 'api'))) + + # check the directories for existance; first one wins + for dir in dirs: + if os.path.isdir(dir): + return dir - - def __getitem__(self, name): - if name in self.messages: - return self.messages[name] return None - def encode(self, msgdef, kwargs): - # Make suitably large buffer - buf = bytearray(self.buffersize) - offset = 0 - size = self.__struct_type(True, msgdef, buf, offset, kwargs) - return buf[:offset + size] - - def decode(self, msgdef, buf): - return self.__struct_type(False, msgdef, buf, 0, None)[1] - - def __struct_type_decode(self, msgdef, buf, offset): - res = [] - off = offset - size = 0 - for k,v in msgdef['args'].iteritems(): - off += size - if type(v) is list: - lst = [] - if callable(v[1]): # compound type - size = 0 - if v[0] in msgdef['args']: # vla - e = res[v[2]] - else: # fixed array - e = v[0] - res.append(lst) - for i in range(e): - (s,l) = v[1](self, False, buf, off + size, None) - lst.append(l) - size += s - continue - if v[1].size == 1: - if type(v[0]) is int: - size = len(buf) - off - else: - size = res[v[2]] - res.append(buf[off:off + size]) - else: - e = v[0] if type(v[0]) is int else res[v[2]] - if e == -1: - e = (len(buf) - off) / v[1].size - lst = [] - res.append(lst) - size = 0 - for i in range(e): - lst.append(v[1].unpack_from(buf, off + size)[0]) - size += v[1].size - else: - if callable(v): - (s,l) = v(self, False, buf, off, None) - res.append(l) - size += s - else: - res.append(v.unpack_from(buf, off)[0]) - size = v.size - - return off + size - offset, msgdef['return_tuple']._make(res) - - def ret_tup(self, name): - if name in self.messages and 'return_tuple' in self.messages[name]: - return self.messages[name]['return_tuple'] - return None - - def add_message(self, name, msgdef): - if name in self.messages: - raise ValueError('Duplicate message name: ' + name) - - args = collections.OrderedDict() - argtypes = collections.OrderedDict() - fields = [] - msg = {} - for i, f in enumerate(msgdef): - if type(f) is dict and 'crc' in f: - msg['crc'] = f['crc'] - continue - field_type = f[0] - field_name = f[1] - if len(f) == 3 and f[2] == 0 and i != len(msgdef) - 2: - raise ValueError('Variable Length Array must be last: ' + name) - args[field_name] = self.__struct(*f) - argtypes[field_name] = field_type - if len(f) == 4: # Find offset to # elements field - args[field_name].append(args.keys().index(f[3]) - i) - fields.append(field_name) - msg['return_tuple'] = collections.namedtuple(name, fields, - rename = True) - self.messages[name] = msg - self.messages[name]['args'] = args - self.messages[name]['argtypes'] = argtypes - return self.messages[name] - - def add_type(self, name, typedef): - return self.add_message('vl_api_' + name + '_t', typedef) - - def make_function(self, name, i, msgdef, multipart, async): - if (async): - f = lambda **kwargs: (self._call_vpp_async(i, msgdef, multipart, **kwargs)) + @classmethod + def find_api_files(cls, api_dir=None, patterns='*'): + """Find API definition files from the given directory tree with the + given pattern. If no directory is given then find_api_dir() is used + to locate one. If no pattern is given then all definition files found + in the directory tree are used. + + :param api_dir: A directory tree in which to locate API definition + files; subdirectories are descended into. + If this is None then find_api_dir() is called to discover it. + :param patterns: A list of patterns to use in each visited directory + when looking for files. + This can be a list/tuple object or a comma-separated string of + patterns. Each value in the list will have leading/trialing + whitespace stripped. + The pattern specifies the first part of the filename, '.api.json' + is appended. + The results are de-duplicated, thus overlapping patterns are fine. + If this is None it defaults to '*' meaning "all API files". + :returns: A list of file paths for the API files found. + """ + if api_dir is None: + api_dir = cls.find_api_dir() + if api_dir is None: + raise VPPApiError("api_dir cannot be located") + + if isinstance(patterns, list) or isinstance(patterns, tuple): + patterns = [p.strip() + '.api.json' for p in patterns] + else: + patterns = [p.strip() + '.api.json' for p in patterns.split(",")] + + api_files = [] + for root, dirnames, files in os.walk(api_dir): + # iterate all given patterns and de-dup the result + files = set(sum([fnmatch.filter(files, p) for p in patterns], [])) + for filename in files: + api_files.append(os.path.join(root, filename)) + + return api_files + + @property + def api(self): + if not hasattr(self, "_api"): + raise VPPApiError("Not connected, api definitions not available") + return self._api + + def make_function(self, msg, i, multipart, do_async): + if (do_async): + def f(**kwargs): + return self._call_vpp_async(i, msg, **kwargs) else: - f = lambda **kwargs: (self._call_vpp(i, msgdef, multipart, **kwargs)) - args = self.messages[name]['args'] - argtypes = self.messages[name]['argtypes'] - f.__name__ = str(name) - f.__doc__ = ", ".join(["%s %s" % (argtypes[k], k) for k in args.keys()]) + def f(**kwargs): + return self._call_vpp(i, msg, multipart, **kwargs) + + f.__name__ = str(msg.name) + f.__doc__ = ", ".join(["%s %s" % + (msg.fieldtypes[j], k) + for j, k in enumerate(msg.fields)]) return f - def _register_functions(self, async=False): + def _register_functions(self, do_async=False): self.id_names = [None] * (self.vpp_dictionary_maxid + 1) self.id_msgdef = [None] * (self.vpp_dictionary_maxid + 1) - for name, msgdef in self.messages.iteritems(): - if name in self.vpp_dictionary: - if self.messages[name]['crc'] != self.vpp_dictionary[name]['crc']: - raise ValueError(3, 'Failed CRC checksum ' + name + - ' ' + self.messages[name]['crc'] + - ' ' + self.vpp_dictionary[name]['crc']) - i = self.vpp_dictionary[name]['id'] - self.id_msgdef[i] = msgdef + self._api = VppApiDynamicMethodHolder() + for name, msg in vpp_iterator(self.messages): + n = name + '_' + msg.crc[2:] + i = self.transport.get_msg_index(n.encode()) + if i > 0: + self.id_msgdef[i] = msg self.id_names[i] = name - multipart = True if name.find('_dump') > 0 else False - setattr(self, name, self.make_function(name, i, msgdef, multipart, async)) - - def _write (self, buf): - if not self.connected: - raise IOError(1, 'Not connected') - return vpp_api.write(str(buf)) - - def _load_dictionary(self): - self.vpp_dictionary = {} - self.vpp_dictionary_maxid = 0 - d = vpp_api.msg_table() - - if not d: - raise IOError(3, 'Cannot get VPP API dictionary') - for i,n in d: - name, crc = n.rsplit('_', 1) - crc = '0x' + crc - self.vpp_dictionary[name] = { 'id' : i, 'crc' : crc } - self.vpp_dictionary_maxid = max(self.vpp_dictionary_maxid, i) - - def connect(self, name, chroot_prefix = None, async = False): - msg_handler = self.msg_handler if not async else self.msg_handler_async - if not chroot_prefix: - rv = vpp_api.connect(name, msg_handler) - else: - rv = vpp_api.connect(name, msg_handler, chroot_prefix) - if rv != 0: - raise IOError(2, 'Connect failed') - self.connected = True + # Create function for client side messages. + if name in self.services: + if 'stream' in self.services[name] and \ + self.services[name]['stream']: + multipart = True + else: + multipart = False + f = self.make_function(msg, i, multipart, do_async) + setattr(self._api, name, FuncWrapper(f)) + else: + self.logger.debug( + 'No such message type or failed CRC checksum: %s', n) - self._load_dictionary() - self._register_functions(async=async) + def connect_internal(self, name, msg_handler, chroot_prefix, rx_qlen, + do_async): + pfx = chroot_prefix.encode() if chroot_prefix else None + + rv = self.transport.connect(name.encode(), pfx, msg_handler, rx_qlen) + if rv != 0: + raise VPPIOError(2, 'Connect failed') + self.vpp_dictionary_maxid = self.transport.msg_table_max_index() + self._register_functions(do_async=do_async) # Initialise control ping - self.control_ping_index = self.vpp_dictionary['control_ping']['id'] + crc = self.messages['control_ping'].crc + self.control_ping_index = self.transport.get_msg_index( + ('control_ping' + '_' + crc[2:]).encode()) self.control_ping_msgdef = self.messages['control_ping'] - - def disconnect(self): - rv = vpp_api.disconnect() + if self.async_thread: + self.event_thread = threading.Thread( + target=self.thread_msg_handler) + self.event_thread.daemon = True + self.event_thread.start() return rv - def results_wait(self, context): - return (self.results[context]['e'].wait(self.timeout)) - - def results_prepare(self, context): - self.results[context] = {} - self.results[context]['e'] = threading.Event() - self.results[context]['e'].clear() - self.results[context]['r'] = [] + def connect(self, name, chroot_prefix=None, do_async=False, rx_qlen=32): + """Attach to VPP. - def results_clean(self, context): - del self.results[context] + name - the name of the client. + chroot_prefix - if VPP is chroot'ed, the prefix of the jail + do_async - if true, messages are sent without waiting for a reply + rx_qlen - the length of the VPP message receive queue between + client and server. + """ + msg_handler = self.transport.get_callback(do_async) + return self.connect_internal(name, msg_handler, chroot_prefix, rx_qlen, + do_async) - def msg_handler(self, msg): - if not msg: - eprint('vpp_api.read failed') - return + def connect_sync(self, name, chroot_prefix=None, rx_qlen=32): + """Attach to VPP in synchronous mode. Application must poll for events. - i, ci = self.header.unpack_from(msg, 0) - if self.id_names[i] == 'rx_thread_exit': - return; + name - the name of the client. + chroot_prefix - if VPP is chroot'ed, the prefix of the jail + rx_qlen - the length of the VPP message receive queue between + client and server. + """ - # - # Decode message and returns a tuple. - # - msgdef = self.id_msgdef[i] - if not msgdef: - raise IOError(2, 'Reply message undefined') + return self.connect_internal(name, None, chroot_prefix, rx_qlen, + do_async=False) - r = self.decode(msgdef, msg) - if 'context' in r._asdict(): - if r.context > 0: - context = r.context + def disconnect(self): + """Detach from VPP.""" + rv = self.transport.disconnect() + self.message_queue.put("terminate event thread") + return rv - msgname = type(r).__name__ + def msg_handler_sync(self, msg): + """Process an incoming message from VPP in sync mode. - # - # XXX: Call provided callback for event - # Are we guaranteed to not get an event during processing of other messages? - # How to differentiate what's a callback message and what not? Context = 0? - # - #if not is_waiting_for_reply(): - if r.context == 0 and self.event_callback: - self.event_callback(msgname, r) + The message may be a reply or it may be an async notification. + """ + r = self.decode_incoming_msg(msg) + if r is None: return - # - # Collect results until control ping - # - if msgname == 'control_ping_reply': - self.results[context]['e'].set() - return + # If we have a context, then use the context to find any + # request waiting for a reply + context = 0 + if hasattr(r, 'context') and r.context > 0: + context = r.context - if not context in self.results: - eprint('Not expecting results for this context', context, r) - return + if context == 0: + # No context -> async notification that we feed to the callback + self.message_queue.put_nowait(r) + else: + raise VPPIOError(2, 'RPC reply message received in event handler') + + def has_context(self, msg): + if len(msg) < 10: + return False + + header = VPPType('header_with_context', [['u16', 'msgid'], + ['u32', 'client_index'], + ['u32', 'context']]) - if 'm' in self.results[context]: - self.results[context]['r'].append(r) + (i, ci, context), size = header.unpack(msg, 0) + if self.id_names[i] == 'rx_thread_exit': return - self.results[context]['r'] = r - self.results[context]['e'].set() + # + # Decode message and returns a tuple. + # + msgobj = self.id_msgdef[i] + if 'context' in msgobj.field_by_name and context >= 0: + return True + return False - def msg_handler_async(self, msg): + def decode_incoming_msg(self, msg, no_type_conversion=False): if not msg: - eprint('vpp_api.read failed') + self.logger.warning('vpp_api.read failed') return - i, ci = self.header.unpack_from(msg, 0) + (i, ci), size = self.header.unpack(msg, 0) if self.id_names[i] == 'rx_thread_exit': - return; + return # # Decode message and returns a tuple. # - msgdef = self.id_msgdef[i] - if not msgdef: - raise IOError(2, 'Reply message undefined') + msgobj = self.id_msgdef[i] + if not msgobj: + raise VPPIOError(2, 'Reply message undefined') + + r, size = msgobj.unpack(msg, ntc=no_type_conversion) + return r + + def msg_handler_async(self, msg): + """Process a message from VPP in async mode. + + In async mode, all messages are returned to the callback. + """ + r = self.decode_incoming_msg(msg) + if r is None: + return - r = self.decode(msgdef, msg) msgname = type(r).__name__ - self.event_callback(msgname, r) + if self.event_callback: + self.event_callback(msgname, r) def _control_ping(self, context): - self._write(self.encode(self.control_ping_msgdef, - { '_vl_msg_id' : self.control_ping_index, - 'context' : context})) + """Send a ping command.""" + self._call_vpp_async(self.control_ping_index, + self.control_ping_msgdef, + context=context) + + def validate_args(self, msg, kwargs): + d = set(kwargs.keys()) - set(msg.field_by_name.keys()) + if d: + raise VPPValueError('Invalid argument {} to {}' + .format(list(d), msg.name)) def _call_vpp(self, i, msgdef, multipart, **kwargs): - if not 'context' in kwargs: + """Given a message, send the message and await a reply. + + msgdef - the message packing definition + i - the message type index + multipart - True if the message returns multiple + messages in return. + context - context number - chosen at random if not + supplied. + The remainder of the kwargs are the arguments to the API call. + + The return value is the message or message array containing + the response. It will raise an IOError exception if there was + no response within the timeout window. + """ + + if 'context' not in kwargs: context = self.get_context() kwargs['context'] = context else: context = kwargs['context'] kwargs['_vl_msg_id'] = i - b = self.encode(msgdef, kwargs) - self.results_prepare(context) - self._write(b) + no_type_conversion = kwargs.pop('_no_type_conversion', False) + + try: + if self.transport.socket_index: + kwargs['client_index'] = self.transport.socket_index + except AttributeError: + pass + self.validate_args(msgdef, kwargs) + + logging.debug(call_logger(msgdef, kwargs)) + + b = msgdef.pack(kwargs) + self.transport.suspend() + + self.transport.write(b) if multipart: - self.results[context]['m'] = True + # Send a ping after the request - we use its response + # to detect that we have seen all results. self._control_ping(context) - self.results_wait(context) - r = self.results[context]['r'] - self.results_clean(context) - return r - def _call_vpp_async(self, i, msgdef, multipart, **kwargs): - if not 'context' in kwargs: + # Block until we get a reply. + rl = [] + while (True): + msg = self.transport.read() + if not msg: + raise VPPIOError(2, 'VPP API client: read failed') + r = self.decode_incoming_msg(msg, no_type_conversion) + msgname = type(r).__name__ + if context not in r or r.context == 0 or context != r.context: + # Message being queued + self.message_queue.put_nowait(r) + continue + + if not multipart: + rl = r + break + if msgname == 'control_ping_reply': + break + + rl.append(r) + + self.transport.resume() + + logger.debug(return_logger(rl)) + return rl + + def _call_vpp_async(self, i, msg, **kwargs): + """Given a message, send the message and await a reply. + + msgdef - the message packing definition + i - the message type index + context - context number - chosen at random if not + supplied. + The remainder of the kwargs are the arguments to the API call. + """ + if 'context' not in kwargs: context = self.get_context() kwargs['context'] = context else: context = kwargs['context'] + try: + if self.transport.socket_index: + kwargs['client_index'] = self.transport.socket_index + except AttributeError: + kwargs['client_index'] = 0 kwargs['_vl_msg_id'] = i - b = self.encode(msgdef, kwargs) + b = msg.pack(kwargs) - self._write(b) + self.transport.write(b) def register_event_callback(self, callback): + """Register a callback for async messages. + + This will be called for async notifications in sync mode, + and all messages in async mode. In sync mode, replies to + requests will not come here. + + callback is a fn(msg_type_name, msg_type) that will be + called when a message comes in. While this function is + executing, note that (a) you are in a background thread and + may wish to use threading.Lock to protect your datastructures, + and (b) message processing from VPP will stop (so if you take + a long while about it you may provoke reply timeouts or cause + VPP to fill the RX buffer). Passing None will disable the + callback. + """ self.event_callback = callback + def thread_msg_handler(self): + """Python thread calling the user registered message handler. + + This is to emulate the old style event callback scheme. Modern + clients should provide their own thread to poll the event + queue. + """ + while True: + r = self.message_queue.get() + if r == "terminate event thread": + break + msgname = type(r).__name__ + if self.event_callback: + self.event_callback(msgname, r) + + +# vim: tabstop=8 expandtab shiftwidth=4 softtabstop=4