from __future__ import print_function
from __future__ import absolute_import
+import ctypes
import sys
+import multiprocessing as mp
import os
import logging
-import collections
-import struct
+import functools
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
-from . vpp_format import VPPFormat
+from . vpp_serializer import VPPType, VPPEnumType, VPPUnionType
+from . vpp_serializer import VPPMessage, vpp_get_type, VPPTypeAlias
+
+logger = logging.getLogger(__name__)
if sys.version[0] == '2':
import Queue as queue
else:
import queue as queue
+__all__ = ('FuncWrapper', 'VPP', 'VppApiDynamicMethodHolder',
+ 'VppEnum', 'VppEnumType',
+ '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):
return t.enum
-# Python3
-# class VppEnum(metaclass=VppEnumType):
-# pass
-class VppEnum:
- __metaclass__ = VppEnumType
+@metaclass(VppEnumType)
+class VppEnum(object):
+ pass
def vpp_atexit(vpp_weakref):
vpp_instance.disconnect()
-def vpp_iterator(d):
- if sys.version[0] == '2':
+if sys.version[0] == '2':
+ def vpp_iterator(d):
return d.iteritems()
- else:
+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
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 '<FuncWrapper(func=<%s(%s)>)>' % (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 VPP():
+class VPPApiClient(object):
"""VPP interface.
This class provides the APIs to VPP. The APIs are loaded
provides a means to register a callback function to receive
these messages in a background thread.
"""
+ apidir = None
+ VPPApiError = VPPApiError
+ VPPRuntimeError = VPPRuntimeError
+ VPPValueError = VPPValueError
+ VPPNotImplementedError = VPPNotImplementedError
+ VPPIOError = VPPIOError
def process_json_file(self, apidef_file):
api = json.load(apidef_file)
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(t[0]):
+ if not vpp_get_type(k):
if v['type'] == 'enum':
try:
VPPEnumType(t[0], t[1:])
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 ValueError('Unresolved type definitions {}'
- .format(unresolved))
+ 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 NotImplementedError:
+ except VPPNotImplementedError:
self.logger.error('Not implemented error for {}'.format(m[0]))
def __init__(self, apifiles=None, testmode=False, async_thread=True,
self.logger = logger
self.messages = {}
+ self.services = {}
self.id_names = []
self.id_msgdef = []
self.header = VPPType('header', [['u16', 'msgid'],
self.message_queue = queue.Queue()
self.read_timeout = read_timeout
self.async_thread = async_thread
+ self.event_thread = None
+ self.testmode = testmode
+ self.use_socket = use_socket
+ self.server_address = server_address
+ self._apifiles = apifiles
if use_socket:
from . vpp_transport_socket import VppTransport
if testmode:
apifiles = []
else:
- raise
+ raise VPPRuntimeError
for file in apifiles:
with open(file) as apidef_file:
# 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)
atexit.register(vpp_atexit, weakref.ref(self))
class ContextId(object):
- """Thread-safe provider of unique context IDs."""
+ """Multiprocessing-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
+ self.context.value += 1
+ return self.context.value
get_context = ContextId()
def get_type(self, name):
: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'])
+ dirs = [cls.apidir] if cls.apidir else []
# 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
# 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
+ # check the directories for existence; first one wins
for dir in dirs:
if os.path.isdir(dir):
return dir
if api_dir is None:
api_dir = cls.find_api_dir()
if api_dir is None:
- raise RuntimeError("api_dir cannot be located")
+ raise VPPApiError("api_dir cannot be located")
if isinstance(patterns, list) or isinstance(patterns, tuple):
patterns = [p.strip() + '.api.json' for p in patterns]
@property
def api(self):
if not hasattr(self, "_api"):
- raise Exception("Not connected, api definitions not available")
+ raise VPPApiError("Not connected, api definitions not available")
return self._api
def make_function(self, msg, i, multipart, do_async):
f.__doc__ = ", ".join(["%s %s" %
(msg.fieldtypes[j], k)
for j, k in enumerate(msg.fields)])
+ f.msg = msg
+
return f
def _register_functions(self, do_async=False):
self._api = VppApiDynamicMethodHolder()
for name, msg in vpp_iterator(self.messages):
n = name + '_' + msg.crc[2:]
- i = self.transport.get_msg_index(n.encode())
+ i = self.transport.get_msg_index(n.encode('utf-8'))
if i > 0:
self.id_msgdef[i] = msg
self.id_names[i] = name
- # TODO: Fix multipart (use services)
- multipart = True if name.find('_dump') > 0 else False
- f = self.make_function(msg, i, multipart, do_async)
- setattr(self._api, name, FuncWrapper(f))
+
+ # 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)
def connect_internal(self, name, msg_handler, chroot_prefix, rx_qlen,
do_async):
- pfx = chroot_prefix.encode() if chroot_prefix else None
+ pfx = chroot_prefix.encode('utf-8') if chroot_prefix else None
- rv = self.transport.connect(name.encode(), pfx, msg_handler, rx_qlen)
+ rv = self.transport.connect(name.encode('utf-8'), pfx,
+ msg_handler, rx_qlen)
if rv != 0:
- raise IOError(2, 'Connect failed')
+ 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:]).encode())
+ ('control_ping' + '_' + crc[2:]).encode('utf-8'))
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):
def disconnect(self):
"""Detach from VPP."""
rv = self.transport.disconnect()
- self.message_queue.put("terminate event thread")
+ if self.event_thread is not None:
+ self.message_queue.put("terminate event thread")
return rv
def msg_handler_sync(self, msg):
# No context -> async notification that we feed to the callback
self.message_queue.put_nowait(r)
else:
- raise IOError(2, 'RPC reply message received in event handler')
+ 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
- def decode_incoming_msg(self, msg):
+ #
+ # 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:
self.logger.warning('vpp_api.read failed')
return
+
(i, ci), size = self.header.unpack(msg, 0)
if self.id_names[i] == 'rx_thread_exit':
return
#
msgobj = self.id_msgdef[i]
if not msgobj:
- raise IOError(2, 'Reply message undefined')
+ raise VPPIOError(2, 'Reply message undefined')
- r, size = msgobj.unpack(msg)
+ r, size = msgobj.unpack(msg, ntc=no_type_conversion)
return r
def msg_handler_async(self, msg):
def validate_args(self, msg, kwargs):
d = set(kwargs.keys()) - set(msg.field_by_name.keys())
if d:
- raise ValueError('Invalid argument {} to {}'
- .format(list(d), msg.name))
+ raise VPPValueError('Invalid argument {} to {}'
+ .format(list(d), msg.name))
- def _call_vpp(self, i, msg, multipart, **kwargs):
+ def _call_vpp(self, i, msgdef, multipart, **kwargs):
"""Given a message, send the message and await a reply.
msgdef - the message packing definition
context = kwargs['context']
kwargs['_vl_msg_id'] = i
+ 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(msg, kwargs)
- b = msg.pack(kwargs)
+ self.validate_args(msgdef, kwargs)
+
+ logging.debug(call_logger(msgdef, kwargs))
+
+ b = msgdef.pack(kwargs)
self.transport.suspend()
self.transport.write(b)
while (True):
msg = self.transport.read()
if not msg:
- raise IOError(2, 'VPP API client: read failed')
- r = self.decode_incoming_msg(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.transport.resume()
+ logger.debug(return_logger(rl))
return rl
def _call_vpp_async(self, i, msg, **kwargs):
if self.event_callback:
self.event_callback(msgname, r)
+ def __repr__(self):
+ return "<VPPApiClient apifiles=%s, testmode=%s, async_thread=%s, " \
+ "logger=%s, read_timeout=%s, use_socket=%s, " \
+ "server_address='%s'>" % (
+ self._apifiles, self.testmode, self.async_thread,
+ self.logger, self.read_timeout, self.use_socket,
+ self.server_address)
+
+
+# Provide the old name for backward compatibility.
+VPP = VPPApiClient
# vim: tabstop=8 expandtab shiftwidth=4 softtabstop=4
Code Review / vpp.git / blobdiff