X-Git-Url: https://gerrit.fd.io/r/gitweb?a=blobdiff_plain;f=src%2Fvpp-api%2Fpython%2Fvpp_papi%2Fvpp_papi.py;h=5a9e0a70c64ed81e9a341da99d66aea6f01945c3;hb=1effb29aa2c3e9179192dd3919c8d849fcb08dae;hp=cefbe37021be8da9cf4c8d308f20e910681a4f43;hpb=1fd5d01eff9d97b8bb34edaaaa7703b5a04b4941;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 cefbe37021b..5a9e0a70c64 100644 --- a/src/vpp-api/python/vpp_papi/vpp_papi.py +++ b/src/vpp-api/python/vpp_papi/vpp_papi.py @@ -1,4 +1,4 @@ -#!/usr/bin/env python +#!/usr/bin/env python3 # # Copyright (c) 2016 Cisco and/or its affiliates. # Licensed under the Apache License, Version 2.0 (the "License"); @@ -15,23 +15,396 @@ # from __future__ import print_function -import sys, os, logging, collections, struct, json, threading, glob +from __future__ import absolute_import +import ctypes +import ipaddress +import sys +import multiprocessing as mp +import os +import queue +import logging +import functools +import json +import threading +import fnmatch +import weakref import atexit +import time +from .vpp_format import verify_enum_hint +from .vpp_serializer import VPPType, VPPEnumType, VPPEnumFlagType, VPPUnionType +from .vpp_serializer import VPPMessage, vpp_get_type, VPPTypeAlias + +try: + import VppTransport +except ModuleNotFoundError: + + class V: + """placeholder for VppTransport as the implementation is dependent on + VPPAPIClient's initialization values + """ + + VppTransport = V + +from .vpp_transport_socket import VppTransport + +logger = logging.getLogger("vpp_papi") +logger.addHandler(logging.NullHandler()) + +__all__ = ( + "FuncWrapper", + "VppApiDynamicMethodHolder", + "VppEnum", + "VppEnumType", + "VppEnumFlag", + "VPPIOError", + "VPPRuntimeError", + "VPPValueError", + "VPPApiClient", +) + + +def metaclass(metaclass): + @functools.wraps(metaclass) + def wrapper(cls): + return metaclass(cls.__name__, cls.__bases__, cls.__dict__.copy()) + + return wrapper + + +class VppEnumType(type): + def __getattr__(cls, name): + t = vpp_get_type(name) + return t.enum -logging.basicConfig(level=logging.DEBUG) -import vpp_api -def eprint(*args, **kwargs): - """Print critical diagnostics to stderr.""" - print(*args, file=sys.stderr, **kwargs) +@metaclass(VppEnumType) +class VppEnum: + pass -def vpp_atexit(self): + +@metaclass(VppEnumType) +class VppEnumFlag: + pass + + +def vpp_atexit(vpp_weakref): """Clean up VPP connection on shutdown.""" - if self.connected: - eprint ('Cleaning up VPP on exit') - self.disconnect() + vpp_instance = vpp_weakref() + if vpp_instance and vpp_instance.transport.connected: + logger.debug("Cleaning up VPP on exit") + vpp_instance.disconnect() + + +def add_convenience_methods(): + # provide convenience methods to IP[46]Address.vapi_af + def _vapi_af(self): + if 6 == self._version: + return VppEnum.vl_api_address_family_t.ADDRESS_IP6.value + if 4 == self._version: + return VppEnum.vl_api_address_family_t.ADDRESS_IP4.value + raise ValueError("Invalid _version.") + + def _vapi_af_name(self): + if 6 == self._version: + return "ip6" + if 4 == self._version: + return "ip4" + raise ValueError("Invalid _version.") + + ipaddress._IPAddressBase.vapi_af = property(_vapi_af) + ipaddress._IPAddressBase.vapi_af_name = property(_vapi_af_name) + + +class VppApiDynamicMethodHolder: + pass + + +class FuncWrapper: + def __init__(self, func): + self._func = func + self.__name__ = func.__name__ + self.__doc__ = func.__doc__ + + def __call__(self, **kwargs): + return self._func(**kwargs) + + def __repr__(self): + return ")>" % (self.__name__, self.__doc__) + + +class VPPApiError(Exception): + pass + + +class VPPNotImplementedError(NotImplementedError): + pass + + +class VPPIOError(IOError): + pass + + +class VPPRuntimeError(RuntimeError): + pass + + +class VPPValueError(ValueError): + pass + + +class VPPApiJSONFiles: + @classmethod + def find_api_dir(cls, dirs=[]): + """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. + """ + + # 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 -class VPP(): + if os.getenv("VPP_API_DIR"): + dirs.append(os.getenv("VPP_API_DIR")) + + if hasattr(main, "__file__"): + # get the path of the calling script + localdir = os.path.dirname(os.path.realpath(main.__file__)) + else: + # 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 existence; first one wins + for dir in dirs: + if os.path.isdir(dir): + return dir + + return None + + @classmethod + def find_api_files(cls, api_dir=None, patterns="*"): # -> list + """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 + + @classmethod + def process_json_file(self, apidef_file): + return self._process_json(apidef_file.read()) + + @classmethod + def process_json_str(self, json_str): + return self._process_json(json_str) + + @staticmethod + def _process_json(json_str): # -> Tuple[Dict, Dict] + api = json.loads(json_str) + types = {} + services = {} + messages = {} + try: + for t in api["enums"]: + t[0] = "vl_api_" + t[0] + "_t" + types[t[0]] = {"type": "enum", "data": t} + except KeyError: + pass + try: + for t in api["enumflags"]: + t[0] = "vl_api_" + t[0] + "_t" + types[t[0]] = {"type": "enum", "data": t} + except KeyError: + pass + try: + for t in api["unions"]: + t[0] = "vl_api_" + t[0] + "_t" + types[t[0]] = {"type": "union", "data": t} + except KeyError: + pass + + try: + for t in api["types"]: + t[0] = "vl_api_" + t[0] + "_t" + types[t[0]] = {"type": "type", "data": t} + except KeyError: + pass + + try: + for t, v in api["aliases"].items(): + types["vl_api_" + t + "_t"] = {"type": "alias", "data": v} + except KeyError: + pass + + try: + services.update(api["services"]) + except KeyError: + pass + + 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 + if not vpp_get_type(k): + if v["type"] == "enumflag": + try: + VPPEnumFlagType(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 + try: + for m in api["messages"]: + try: + messages[m[0]] = VPPMessage(m[0], m[1:]) + except VPPNotImplementedError: + ### OLE FIXME + logger.error("Not implemented error for {}".format(m[0])) + except KeyError: + pass + return messages, services + + @staticmethod + def load_api(apifiles=None, apidir=None): + messages = {} + services = {} + if not apifiles: + # Pick up API definitions from default directory + try: + if isinstance(apidir, list): + apifiles = [] + for d in apidir: + apifiles += VPPApiJSONFiles.find_api_files(d) + else: + apifiles = VPPApiJSONFiles.find_api_files(apidir) + except (RuntimeError, VPPApiError): + raise VPPRuntimeError + + for file in apifiles: + with open(file) as apidef_file: + m, s = VPPApiJSONFiles.process_json_file(apidef_file) + messages.update(m) + services.update(s) + + return apifiles, messages, services + + +class VPPApiClient: """VPP interface. This class provides the APIs to VPP. The APIs are loaded @@ -43,7 +416,26 @@ class VPP(): provides a means to register a callback function to receive these messages in a background thread. """ - def __init__(self, apifiles = None, testmode = False): + + VPPApiError = VPPApiError + VPPRuntimeError = VPPRuntimeError + VPPValueError = VPPValueError + VPPNotImplementedError = VPPNotImplementedError + VPPIOError = VPPIOError + + def __init__( + self, + *, + apifiles=None, + apidir=None, + testmode=False, + async_thread=True, + logger=None, + loglevel=None, + read_timeout=5, + use_socket=True, + server_address="/run/vpp/api.sock", + ): """Create a VPP API object. apifiles is a list of files containing API @@ -51,379 +443,191 @@ class VPP(): 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( + "{}.{}".format(__name__, self.__class__.__name__) + ) + if loglevel is not None: + logger.setLevel(loglevel) + self.logger = logger + self.messages = {} + self.services = {} self.id_names = [] self.id_msgdef = [] - self.buffersize = 10000 - self.connected = False - self.header = struct.Struct('>HI') - self.results_lock = threading.Lock() - self.results = {} - self.timeout = 5 + self.header = VPPType("header", [["u16", "msgid"], ["u32", "client_index"]]) self.apifiles = [] - - if not apifiles: - # Pick up API definitions from default directory - apifiles = glob.glob('/usr/share/vpp/api/*.api.json') - - for file in apifiles: - with open(file) as apidef_file: - api = json.load(apidef_file) - for t in api['types']: - self.add_type(t[0], t[1:]) - - for m in api['messages']: - self.add_message(m[0], m[1:]) - self.apifiles = apifiles + self.apidir = apidir + self.event_callback = None + self.message_queue = queue.Queue() + self.read_timeout = read_timeout + self.async_thread = async_thread + self.event_thread = None + self.testmode = testmode + self.server_address = server_address + self._apifiles = apifiles + self.stats = {} + + if self.apidir is None and hasattr(self.__class__, "apidir"): + # Keep supporting the old style of providing apidir. + self.apidir = self.__class__.apidir + try: + self.apifiles, self.messages, self.services = VPPApiJSONFiles.load_api( + apifiles, self.apidir + ) + except VPPRuntimeError as e: + if testmode: + self.apifiles = [] + else: + raise e # 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") + if not (verify_enum_hint(VppEnum.vl_api_address_family_t)): + raise VPPRuntimeError("Invalid address family hints. " "Cannot continue.") + 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, self) + atexit.register(vpp_atexit, weakref.ref(self)) + + add_convenience_methods() + + def get_function(self, name): + return getattr(self._api, name) + + class ContextId: + """Multiprocessing-safe provider of unique context IDs.""" - class ContextId(object): - """Thread-safe provider of unique context IDs.""" def __init__(self): - self.context = 0 - self.lock = threading.Lock() + self.context = mp.Value(ctypes.c_uint, 0) + self.lock = mp.Lock() + def __call__(self): """Get a new unique (or, at least, not recently used) context.""" - with self.lock: - self.context += 1 - return self.context + with self.lock: + self.context.value += 1 + return self.context.value + get_context = ContextId() - def status(self): - """Debug function: report current VPP API status to stdout.""" - print('Connected') if self.connected else print('Not Connected') - print('Read API definitions from', ', '.join(self.apifiles)) - - def __struct (self, t, n = None, e = -1, vl = None): - """Create a packing structure for a message.""" - 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) - ) + def get_type(self, name): + return vpp_get_type(name) - raise ValueError(1, 'Invalid message type: ' + t) + @property + def api(self): + if not hasattr(self, "_api"): + raise VPPApiError("Not connected, api definitions not available") + return self._api - def __struct_type(self, encode, msgdef, buf, offset, kwargs): - """Get a message packer or unpacker.""" - if encode: - return self.__struct_type_encode(msgdef, buf, offset, kwargs) - 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 + def make_function(self, msg, i, multipart, do_async): + if do_async: - return off + size - offset + def f(**kwargs): + return self._call_vpp_async(i, msg, **kwargs) + else: - def __getitem__(self, name): - if name in self.messages: - return self.messages[name] - return None + def f(**kwargs): + return self._call_vpp(i, msg, multipart, **kwargs) - 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 + f.__name__ = str(msg.name) + f.__doc__ = ", ".join( + ["%s %s" % (msg.fieldtypes[j], k) for j, k in enumerate(msg.fields)] + ) + f.msg = msg - return off + size - offset, msgdef['return_tuple']._make(res) + return f - 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 make_pack_function(self, msg, i, multipart): + def f(**kwargs): + return self._call_vpp_pack(i, msg, **kwargs) - 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, **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()]) + f.msg = msg 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 self.messages.items(): + n = name + "_" + msg.crc[2:] + i = self.transport.get_msg_index(n) + 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): - """Send a binary-packed message to VPP.""" - 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, rx_qlen = 32): + + # Create function for client side messages. + if name in self.services: + f = self.make_function(msg, i, self.services[name], do_async) + f_pack = self.make_pack_function(msg, i, self.services[name]) + setattr(self._api, name, FuncWrapper(f)) + setattr(self._api, name + "_pack", FuncWrapper(f_pack)) + else: + self.logger.debug("No such message type or failed CRC checksum: %s", n) + + def connect_internal(self, name, msg_handler, chroot_prefix, rx_qlen, do_async): + pfx = chroot_prefix.encode("utf-8") if chroot_prefix else None + + rv = self.transport.connect(name, pfx, msg_handler, rx_qlen, do_async) + 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 + crc = self.messages["control_ping"].crc + self.control_ping_index = self.transport.get_msg_index( + ("control_ping" + "_" + crc[2:]) + ) + self.control_ping_msgdef = self.messages["control_ping"] + if self.async_thread: + self.event_thread = threading.Thread(target=self.thread_msg_handler) + self.event_thread.daemon = True + self.event_thread.start() + else: + self.event_thread = None + return rv + + def connect(self, name, chroot_prefix=None, do_async=False, rx_qlen=32): """Attach to VPP. name - the name of the client. chroot_prefix - if VPP is chroot'ed, the prefix of the jail - async - if true, messages are sent without waiting for a reply + 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.msg_handler_sync if not async else self.msg_handler_async - if chroot_prefix is not None: - rv = vpp_api.connect(name, msg_handler, rx_qlen, chroot_prefix) - else: - rv = vpp_api.connect(name, msg_handler, rx_qlen) + msg_handler = self.transport.get_callback(do_async) + return self.connect_internal( + name, msg_handler, chroot_prefix, rx_qlen, do_async + ) - if rv != 0: - raise IOError(2, 'Connect failed') - self.connected = True + def connect_sync(self, name, chroot_prefix=None, rx_qlen=32): + """Attach to VPP in synchronous mode. Application must poll for events. - self._load_dictionary() - self._register_functions(async=async) + 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. + """ - # Initialise control ping - self.control_ping_index = self.vpp_dictionary['control_ping']['id'] - self.control_ping_msgdef = self.messages['control_ping'] + return self.connect_internal(name, None, chroot_prefix, rx_qlen, do_async=False) def disconnect(self): """Detach from VPP.""" - rv = vpp_api.disconnect() - self.connected = False + rv = self.transport.disconnect() + if self.event_thread is not None: + self.message_queue.put("terminate event thread") return rv - def results_wait(self, context): - """In a sync call, wait for the reply - - The context ID is used to pair reply to request. - """ - - # Results is filled by the background callback. It will - # raise the event when the context receives a response. - # Given there are two threads we have to be careful with the - # use of results and the structures under it, hence the lock. - with self.results_lock: - result = self.results[context] - ev = result['e'] - - timed_out = not ev.wait(self.timeout) - - if timed_out: - raise IOError(3, 'Waiting for reply timed out') - else: - with self.results_lock: - result = self.results[context] - del self.results[context] - return result['r'] - - def results_prepare(self, context, multi=False): - """Prep for receiving a result in response to a request msg - - context - unique context number sent in request and - returned in reply or replies - multi - true if we expect multiple messages from this - reply. - """ - - # The event is used to indicate that all results are in - new_result = { - 'e': threading.Event(), - } - if multi: - # Make it clear to the BG thread it's going to see several - # messages; messages are stored in a results array - new_result['m'] = True - new_result['r'] = [] - - new_result['e'].clear() - - # Put the prepped result structure into results, at which point - # the bg thread can also access it (hence the thread lock) - with self.results_lock: - self.results[context] = new_result - def msg_handler_sync(self, msg): """Process an incoming message from VPP in sync mode. @@ -436,58 +640,53 @@ class VPP(): # 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: + if hasattr(r, "context") and r.context > 0: context = r.context - msgname = type(r).__name__ - if context == 0: # No context -> async notification that we feed to the callback - if self.event_callback: - self.event_callback(msgname, r) + self.message_queue.put_nowait(r) else: - # Context -> use the results structure (carefully) to find - # who we're responding to and return the message to that - # thread - with self.results_lock: - if context not in self.results: - eprint('Not expecting results for this context', context, r) - else: - result = self.results[context] - - # - # Collect results until control ping - # - - if msgname == 'control_ping_reply': - # End of a multipart - result['e'].set() - elif 'm' in self.results[context]: - # One element in a multipart - result['r'].append(r) - else: - # All of a single result - result['r'] = r - result['e'].set() - - def decode_incoming_msg(self, msg): + 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"]], + ) + + (i, ci, context), size = header.unpack(msg, 0) + if self.id_names[i] == "rx_thread_exit": + return + + # + # 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 decode_incoming_msg(self, msg, no_type_conversion=False): if not msg: - eprint('vpp_api.read failed') + logger.warning("vpp_api.read failed") return - i, ci = self.header.unpack_from(msg, 0) - if self.id_names[i] == 'rx_thread_exit': + (i, ci), size = self.header.unpack(msg, 0) + if self.id_names[i] == "rx_thread_exit": return # # Decode message and returns a tuple. # - msgdef = self.id_msgdef[i] - if not msgdef: - raise IOError(2, 'Reply message undefined') - - r = self.decode(msgdef, msg) + 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): @@ -501,16 +700,47 @@ class VPP(): msgname = type(r).__name__ - if self.event_callback: - self.event_callback(msgname, r) + if self.event_callback: + self.event_callback(msgname, r) def _control_ping(self, context): """Send a ping command.""" - self._call_vpp_async(self.control_ping_index, - self.control_ping_msgdef, - context=context) + 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 _add_stat(self, name, ms): + if not name in self.stats: + self.stats[name] = {"max": ms, "count": 1, "avg": ms} + else: + if ms > self.stats[name]["max"]: + self.stats[name]["max"] = ms + self.stats[name]["count"] += 1 + n = self.stats[name]["count"] + self.stats[name]["avg"] = self.stats[name]["avg"] * (n - 1) / n + ms / n + + def get_stats(self): + s = "\n=== API PAPI STATISTICS ===\n" + s += "{:<30} {:>4} {:>6} {:>6}\n".format("message", "cnt", "avg", "max") + for n in sorted(self.stats.items(), key=lambda v: v[1]["avg"], reverse=True): + s += "{:<30} {:>4} {:>6.2f} {:>6.2f}\n".format( + n[0], n[1]["count"], n[1]["avg"], n[1]["max"] + ) + return s + + def get_field_options(self, msg, fld_name): + # when there is an option, the msgdef has 3 elements. + # ['u32', 'ring_size', {'default': 1024}] + for _def in self.messages[msg].msgdef: + if isinstance(_def, list) and len(_def) == 3 and _def[1] == fld_name: + return _def[2] - def _call_vpp(self, i, msgdef, multipart, **kwargs): + def _call_vpp(self, i, msgdef, service, **kwargs): """Given a message, send the message and await a reply. msgdef - the message packing definition @@ -525,46 +755,146 @@ class VPP(): the response. It will raise an IOError exception if there was no response within the timeout window. """ - - # We need a context if not supplied, in order to get the - # response - context = kwargs.get('context', self.get_context()) - kwargs['context'] = context - - # Set up to receive a response - self.results_prepare(context, multi=multipart) - - # Output the message - self._call_vpp_async(i, msgdef, **kwargs) - - if multipart: - # Send a ping after the request - we use its response - # to detect that we have seen all results. - self._control_ping(context) + ts = time.time() + if "context" not in kwargs: + context = self.get_context() + kwargs["context"] = context + else: + context = kwargs["context"] + kwargs["_vl_msg_id"] = i + + no_type_conversion = kwargs.pop("_no_type_conversion", False) + timeout = kwargs.pop("_timeout", None) + + try: + if self.transport.socket_index: + kwargs["client_index"] = self.transport.socket_index + except AttributeError: + pass + self.validate_args(msgdef, kwargs) + + s = "Calling {}({})".format( + msgdef.name, ",".join(["{!r}:{!r}".format(k, v) for k, v in kwargs.items()]) + ) + self.logger.debug(s) + + b = msgdef.pack(kwargs) + self.transport.suspend() + + self.transport.write(b) + + msgreply = service["reply"] + stream = True if "stream" in service else False + if stream: + if "stream_msg" in service: + # New service['reply'] = _reply and service['stream_message'] = _details + stream_message = service["stream_msg"] + modern = True + else: + # Old service['reply'] = _details + stream_message = msgreply + msgreply = "control_ping_reply" + modern = False + # Send a ping after the request - we use its response + # to detect that we have seen all results. + self._control_ping(context) # Block until we get a reply. - r = self.results_wait(context) - - return r - - def _call_vpp_async(self, i, msgdef, **kwargs): - """Given a message, send the message and await a reply. + rl = [] + while True: + r = self.read_blocking(no_type_conversion, timeout) + if r is None: + raise VPPIOError(2, "VPP API client: read failed") + 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 msgname != msgreply and (stream and (msgname != stream_message)): + print("REPLY MISMATCH", msgreply, msgname, stream_message, stream) + if not stream: + rl = r + break + if msgname == msgreply: + if modern: # Return both reply and list + rl = r, rl + break + + rl.append(r) + + self.transport.resume() + + s = "Return value: {!r}".format(r) + if len(s) > 80: + s = s[:80] + "..." + self.logger.debug(s) + te = time.time() + self._add_stat(msgdef.name, (te - ts) * 1000) + return rl + + def _call_vpp_async(self, i, msg, **kwargs): + """Given a message, send the message and return the context. 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. + + The reply message(s) will be delivered later to the registered callback. + The returned context will help with assigning which call + the reply belongs to. """ - if not 'context' in kwargs: + if "context" not in kwargs: context = self.get_context() - kwargs['context'] = context + kwargs["context"] = context else: - context = kwargs['context'] - kwargs['_vl_msg_id'] = i - b = self.encode(msgdef, kwargs) + 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 = msg.pack(kwargs) + + self.transport.write(b) + return context + + def _call_vpp_pack(self, i, msg, **kwargs): + """Given a message, return the binary representation.""" + kwargs["_vl_msg_id"] = i + kwargs["client_index"] = 0 + kwargs["context"] = 0 + return msg.pack(kwargs) + + def read_blocking(self, no_type_conversion=False, timeout=None): + """Get next received message from transport within timeout, decoded. + + Note that notifications have context zero + and are not put into receive queue (at least for socket transport), + use async_thread with registered callback for processing them. + + If no message appears in the queue within timeout, return None. + + Optionally, type conversion can be skipped, + as some of conversions are into less precise types. + + When r is the return value of this, the caller can get message name as: + msgname = type(r).__name__ + and context number (type long) as: + context = r.context - self._write(b) + :param no_type_conversion: If false, type conversions are applied. + :type no_type_conversion: bool + :returns: Decoded message, or None if no message (within timeout). + :rtype: Whatever VPPType.unpack returns, depends on no_type_conversion. + :raises VppTransportShmemIOError if timed out. + """ + msg = self.transport.read(timeout=timeout) + if not msg: + return None + return self.decode_incoming_msg(msg, no_type_conversion) def register_event_callback(self, callback): """Register a callback for async messages. @@ -583,3 +913,72 @@ class VPP(): 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) + + def validate_message_table(self, namecrctable): + """Take a dictionary of name_crc message names + and returns an array of missing messages""" + + missing_table = [] + for name_crc in namecrctable: + i = self.transport.get_msg_index(name_crc) + if i <= 0: + missing_table.append(name_crc) + return missing_table + + def dump_message_table(self): + """Return VPPs API message table as name_crc dictionary""" + return self.transport.message_table + + def dump_message_table_filtered(self, msglist): + """Return VPPs API message table as name_crc dictionary, + filtered by message name list.""" + + replies = [self.services[n]["reply"] for n in msglist] + message_table_filtered = {} + for name in msglist + replies: + for k, v in self.transport.message_table.items(): + if k.startswith(name): + message_table_filtered[k] = v + break + return message_table_filtered + + def __repr__(self): + return ( + "" + % ( + self._apifiles, + self.testmode, + self.async_thread, + self.logger, + self.read_timeout, + self.server_address, + ) + ) + + def details_iter(self, f, **kwargs): + cursor = 0 + while True: + kwargs["cursor"] = cursor + rv, details = f(**kwargs) + for d in details: + yield d + if rv.retval == 0 or rv.retval != -165: + break + cursor = rv.cursor