src/vpp-api/python/vpp_papi/vpp_papi.py

   1 #!/usr/bin/env python
   2 #
   3 # Copyright (c) 2016 Cisco and/or its affiliates.
   4 # Licensed under the Apache License, Version 2.0 (the "License");
   5 # you may not use this file except in compliance with the License.
   6 # You may obtain a copy of the License at:
   7 #
   8 #     http://www.apache.org/licenses/LICENSE-2.0
   9 #
  10 # Unless required by applicable law or agreed to in writing, software
  11 # distributed under the License is distributed on an "AS IS" BASIS,
  12 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13 # See the License for the specific language governing permissions and
  14 # limitations under the License.
  15 #
  16
  17 from __future__ import print_function
  18 import sys, os, logging, collections, struct, json, threading, glob
  19 import atexit
  20
  21 logging.basicConfig(level=logging.DEBUG)
  22 import vpp_api
  23
  24 def eprint(*args, **kwargs):
  25     """Print critical diagnostics to stderr."""
  26     print(*args, file=sys.stderr, **kwargs)
  27
  28 def vpp_atexit(self):
  29     """Clean up VPP connection on shutdown."""
  30     if self.connected:
  31         eprint ('Cleaning up VPP on exit')
  32         self.disconnect()
  33
  34
  35 class Empty(object):
  36     pass
  37
  38
  39 class FuncWrapper(object):
  40     def __init__(self, func):
  41         self._func = func
  42         self.__name__ = func.__name__
  43
  44     def __call__(self, **kwargs):
  45         return self._func(**kwargs)
  46
  47
  48 class VPP():
  49     """VPP interface.
  50
  51     This class provides the APIs to VPP.  The APIs are loaded
  52     from provided .api.json files and makes functions accordingly.
  53     These functions are documented in the VPP .api files, as they
  54     are dynamically created.
  55
  56     Additionally, VPP can send callback messages; this class
  57     provides a means to register a callback function to receive
  58     these messages in a background thread.
  59     """
  60     def __init__(self, apifiles = None, testmode = False):
  61         """Create a VPP API object.
  62
  63         apifiles is a list of files containing API
  64         descriptions that will be loaded - methods will be
  65         dynamically created reflecting these APIs.  If not
  66         provided this will load the API files from VPP's
  67         default install location.
  68         """
  69         self.messages = {}
  70         self.id_names = []
  71         self.id_msgdef = []
  72         self.buffersize = 10000
  73         self.connected = False
  74         self.header = struct.Struct('>HI')
  75         self.results_lock = threading.Lock()
  76         self.results = {}
  77         self.timeout = 5
  78         self.apifiles = []
  79         self.event_callback = None
  80
  81         if not apifiles:
  82             # Pick up API definitions from default directory
  83             apifiles = glob.glob('/usr/share/vpp/api/*.api.json')
  84
  85         for file in apifiles:
  86             with open(file) as apidef_file:
  87                 api = json.load(apidef_file)
  88                 for t in api['types']:
  89                     self.add_type(t[0], t[1:])
  90
  91                 for m in api['messages']:
  92                     self.add_message(m[0], m[1:])
  93         self.apifiles = apifiles
  94
  95         # Basic sanity check
  96         if len(self.messages) == 0 and not testmode:
  97             raise ValueError(1, 'Missing JSON message definitions')
  98
  99         # Make sure we allow VPP to clean up the message rings.
 100         atexit.register(vpp_atexit, self)
 101
 102     class ContextId(object):
 103         """Thread-safe provider of unique context IDs."""
 104         def __init__(self):
 105             self.context = 0
 106             self.lock = threading.Lock()
 107         def __call__(self):
 108             """Get a new unique (or, at least, not recently used) context."""
 109             with self.lock:
 110                 self.context += 1
 111                 return self.context
 112     get_context = ContextId()
 113
 114     def status(self):
 115         """Debug function: report current VPP API status to stdout."""
 116         print('Connected') if self.connected else print('Not Connected')
 117         print('Read API definitions from', ', '.join(self.apifiles))
 118
 119     def __struct (self, t, n = None, e = -1, vl = None):
 120         """Create a packing structure for a message."""
 121         base_types = { 'u8' : 'B',
 122                        'u16' : 'H',
 123                        'u32' : 'I',
 124                        'i32' : 'i',
 125                        'u64' : 'Q',
 126                        'f64' : 'd',
 127                        }
 128         pack = None
 129         if t in base_types:
 130             pack = base_types[t]
 131             if not vl:
 132                 if e > 0 and t == 'u8':
 133                     # Fixed byte array
 134                     return struct.Struct('>' + str(e) + 's')
 135                 if e > 0:
 136                     # Fixed array of base type
 137                     return [e, struct.Struct('>' + base_types[t])]
 138                 elif e == 0:
 139                     # Old style variable array
 140                     return [-1, struct.Struct('>' + base_types[t])]
 141             else:
 142                 # Variable length array
 143                 return [vl, struct.Struct('>s')] if t == 'u8' else \
 144                     [vl, struct.Struct('>' + base_types[t])]
 145
 146             return struct.Struct('>' + base_types[t])
 147
 148         if t in self.messages:
 149             ### Return a list in case of array ###
 150             if e > 0 and not vl:
 151                 return [e, lambda self, encode, buf, offset, args: (
 152                     self.__struct_type(encode, self.messages[t], buf, offset,
 153                                        args))]
 154             if vl:
 155                 return [vl, lambda self, encode, buf, offset, args: (
 156                     self.__struct_type(encode, self.messages[t], buf, offset,
 157                                        args))]
 158             elif e == 0:
 159                 # Old style VLA
 160                 raise NotImplementedError(1, 'No support for compound types ' + t)
 161             return lambda self, encode, buf, offset, args: (
 162                 self.__struct_type(encode, self.messages[t], buf, offset, args)
 163             )
 164
 165         raise ValueError(1, 'Invalid message type: ' + t)
 166
 167     def __struct_type(self, encode, msgdef, buf, offset, kwargs):
 168         """Get a message packer or unpacker."""
 169         if encode:
 170             return self.__struct_type_encode(msgdef, buf, offset, kwargs)
 171         else:
 172             return self.__struct_type_decode(msgdef, buf, offset)
 173
 174     def __struct_type_encode(self, msgdef, buf, offset, kwargs):
 175         off = offset
 176         size = 0
 177
 178         for k in kwargs:
 179             if k not in msgdef['args']:
 180                 raise ValueError(1, 'Invalid field-name in message call ' + k)
 181
 182         for k,v in msgdef['args'].iteritems():
 183             off += size
 184             if k in kwargs:
 185                 if type(v) is list:
 186                     if callable(v[1]):
 187                         e = kwargs[v[0]] if v[0] in kwargs else v[0]
 188                         size = 0
 189                         for i in range(e):
 190                             size += v[1](self, True, buf, off + size,
 191                                          kwargs[k][i])
 192                     else:
 193                         if v[0] in kwargs:
 194                             l = kwargs[v[0]]
 195                         else:
 196                             l = len(kwargs[k])
 197                         if v[1].size == 1:
 198                             buf[off:off + l] = bytearray(kwargs[k])
 199                             size = l
 200                         else:
 201                             size = 0
 202                             for i in kwargs[k]:
 203                                 v[1].pack_into(buf, off + size, i)
 204                                 size += v[1].size
 205                 else:
 206                     if callable(v):
 207                         size = v(self, True, buf, off, kwargs[k])
 208                     else:
 209                         v.pack_into(buf, off, kwargs[k])
 210                         size = v.size
 211             else:
 212                 size = v.size if not type(v) is list else 0
 213
 214         return off + size - offset
 215
 216
 217     def __getitem__(self, name):
 218         if name in self.messages:
 219             return self.messages[name]
 220         return None
 221
 222     def encode(self, msgdef, kwargs):
 223         # Make suitably large buffer
 224         buf = bytearray(self.buffersize)
 225         offset = 0
 226         size = self.__struct_type(True, msgdef, buf, offset, kwargs)
 227         return buf[:offset + size]
 228
 229     def decode(self, msgdef, buf):
 230         return self.__struct_type(False, msgdef, buf, 0, None)[1]
 231
 232     def __struct_type_decode(self, msgdef, buf, offset):
 233         res = []
 234         off = offset
 235         size = 0
 236         for k,v in msgdef['args'].iteritems():
 237             off += size
 238             if type(v) is list:
 239                 lst = []
 240                 if callable(v[1]): # compound type
 241                     size = 0
 242                     if v[0] in msgdef['args']: # vla
 243                         e = res[v[2]]
 244                     else: # fixed array
 245                         e = v[0]
 246                     res.append(lst)
 247                     for i in range(e):
 248                         (s,l) = v[1](self, False, buf, off + size, None)
 249                         lst.append(l)
 250                         size += s
 251                     continue
 252                 if v[1].size == 1:
 253                     if type(v[0]) is int:
 254                         size = len(buf) - off
 255                     else:
 256                         size = res[v[2]]
 257                     res.append(buf[off:off + size])
 258                 else:
 259                     e = v[0] if type(v[0]) is int else res[v[2]]
 260                     if e == -1:
 261                         e = (len(buf) - off) / v[1].size
 262                     lst = []
 263                     res.append(lst)
 264                     size = 0
 265                     for i in range(e):
 266                         lst.append(v[1].unpack_from(buf, off + size)[0])
 267                         size += v[1].size
 268             else:
 269                 if callable(v):
 270                     (s,l) = v(self, False, buf, off, None)
 271                     res.append(l)
 272                     size += s
 273                 else:
 274                     res.append(v.unpack_from(buf, off)[0])
 275                     size = v.size
 276
 277         return off + size - offset, msgdef['return_tuple']._make(res)
 278
 279     def ret_tup(self, name):
 280         if name in self.messages and 'return_tuple' in self.messages[name]:
 281             return self.messages[name]['return_tuple']
 282         return None
 283
 284     def add_message(self, name, msgdef):
 285         if name in self.messages:
 286             raise ValueError('Duplicate message name: ' + name)
 287
 288         args = collections.OrderedDict()
 289         argtypes = collections.OrderedDict()
 290         fields = []
 291         msg = {}
 292         for i, f in enumerate(msgdef):
 293             if type(f) is dict and 'crc' in f:
 294                 msg['crc'] = f['crc']
 295                 continue
 296             field_type = f[0]
 297             field_name = f[1]
 298             if len(f) == 3 and f[2] == 0 and i != len(msgdef) - 2:
 299                 raise ValueError('Variable Length Array must be last: ' + name)
 300             args[field_name] = self.__struct(*f)
 301             argtypes[field_name] = field_type
 302             if len(f) == 4: # Find offset to # elements field
 303                 args[field_name].append(args.keys().index(f[3]) - i)
 304             fields.append(field_name)
 305         msg['return_tuple'] = collections.namedtuple(name, fields,
 306                                                      rename = True)
 307         self.messages[name] = msg
 308         self.messages[name]['args'] = args
 309         self.messages[name]['argtypes'] = argtypes
 310         return self.messages[name]
 311
 312     def add_type(self, name, typedef):
 313         return self.add_message('vl_api_' + name + '_t', typedef)
 314
 315     def make_function(self, name, i, msgdef, multipart, async):
 316         if (async):
 317             f = lambda **kwargs: (self._call_vpp_async(i, msgdef, **kwargs))
 318         else:
 319             f = lambda **kwargs: (self._call_vpp(i, msgdef, multipart, **kwargs))
 320         args = self.messages[name]['args']
 321         argtypes = self.messages[name]['argtypes']
 322         f.__name__ = str(name)
 323         f.__doc__ = ", ".join(["%s %s" % (argtypes[k], k) for k in args.keys()])
 324         return f
 325
 326     @property
 327     def api(self):
 328         if not hasattr(self, "_api"):
 329             raise Exception("Not connected, api definitions not available")
 330         return self._api
 331
 332     def _register_functions(self, async=False):
 333         self.id_names = [None] * (self.vpp_dictionary_maxid + 1)
 334         self.id_msgdef = [None] * (self.vpp_dictionary_maxid + 1)
 335         self._api = Empty()
 336         for name, msgdef in self.messages.iteritems():
 337             if name in self.vpp_dictionary:
 338                 if self.messages[name]['crc'] != self.vpp_dictionary[name]['crc']:
 339                     raise ValueError(3, 'Failed CRC checksum ' + name +
 340                                      ' ' + self.messages[name]['crc'] +
 341                                      ' ' + self.vpp_dictionary[name]['crc'])
 342                 i = self.vpp_dictionary[name]['id']
 343                 self.id_msgdef[i] = msgdef
 344                 self.id_names[i] = name
 345                 multipart = True if name.find('_dump') > 0 else False
 346                 f = self.make_function(name, i, msgdef, multipart, async)
 347                 setattr(self._api, name, FuncWrapper(f))
 348
 349                 # olf API stuff starts here - will be removed in 17.07
 350                 if hasattr(self, name):
 351                     raise NameError(
 352                         3, "Conflicting name in JSON definition: `%s'" % name)
 353                 setattr(self, name, f)
 354                 # old API stuff ends here
 355
 356     def _write (self, buf):
 357         """Send a binary-packed message to VPP."""
 358         if not self.connected:
 359             raise IOError(1, 'Not connected')
 360         return vpp_api.write(str(buf))
 361
 362     def _load_dictionary(self):
 363         self.vpp_dictionary = {}
 364         self.vpp_dictionary_maxid = 0
 365         d = vpp_api.msg_table()
 366
 367         if not d:
 368             raise IOError(3, 'Cannot get VPP API dictionary')
 369         for i,n in d:
 370             name, crc =  n.rsplit('_', 1)
 371             crc = '0x' + crc
 372             self.vpp_dictionary[name] = { 'id' : i, 'crc' : crc }
 373             self.vpp_dictionary_maxid = max(self.vpp_dictionary_maxid, i)
 374
 375     def connect(self, name, chroot_prefix = None, async = False, rx_qlen = 32):
 376         """Attach to VPP.
 377
 378         name - the name of the client.
 379         chroot_prefix - if VPP is chroot'ed, the prefix of the jail
 380         async - if true, messages are sent without waiting for a reply
 381         rx_qlen - the length of the VPP message receive queue between
 382         client and server.
 383         """
 384         msg_handler = self.msg_handler_sync if not async else self.msg_handler_async
 385         if chroot_prefix is not None:
 386             rv = vpp_api.connect(name, msg_handler, rx_qlen, chroot_prefix)
 387         else:
 388             rv = vpp_api.connect(name, msg_handler, rx_qlen)
 389
 390         if rv != 0:
 391             raise IOError(2, 'Connect failed')
 392         self.connected = True
 393
 394         self._load_dictionary()
 395         self._register_functions(async=async)
 396
 397         # Initialise control ping
 398         self.control_ping_index = self.vpp_dictionary['control_ping']['id']
 399         self.control_ping_msgdef = self.messages['control_ping']
 400
 401     def disconnect(self):
 402         """Detach from VPP."""
 403         rv = vpp_api.disconnect()
 404         self.connected = False
 405         return rv
 406
 407     def results_wait(self, context):
 408         """In a sync call, wait for the reply
 409
 410         The context ID is used to pair reply to request.
 411         """
 412
 413         # Results is filled by the background callback.  It will
 414         # raise the event when the context receives a response.
 415         # Given there are two threads we have to be careful with the
 416         # use of results and the structures under it, hence the lock.
 417         with self.results_lock:
 418             result = self.results[context]
 419             ev = result['e']
 420
 421         timed_out = not ev.wait(self.timeout)
 422
 423         if timed_out:
 424            raise IOError(3, 'Waiting for reply timed out')
 425         else:
 426             with self.results_lock:
 427                 result = self.results[context]
 428                 del self.results[context]
 429                 return result['r']
 430
 431     def results_prepare(self, context, multi=False):
 432         """Prep for receiving a result in response to a request msg
 433
 434         context - unique context number sent in request and
 435         returned in reply or replies
 436         multi - true if we expect multiple messages from this
 437         reply.
 438         """
 439
 440         # The event is used to indicate that all results are in
 441         new_result = {
 442             'e': threading.Event(),
 443         }
 444         if multi:
 445             # Make it clear to the BG thread it's going to see several
 446             # messages; messages are stored in a results array
 447             new_result['m'] = True
 448             new_result['r'] = []
 449
 450         new_result['e'].clear()
 451
 452         # Put the prepped result structure into results, at which point
 453         # the bg thread can also access it (hence the thread lock)
 454         with self.results_lock:
 455             self.results[context] = new_result
 456
 457     def msg_handler_sync(self, msg):
 458         """Process an incoming message from VPP in sync mode.
 459
 460         The message may be a reply or it may be an async notification.
 461         """
 462         r = self.decode_incoming_msg(msg)
 463         if r is None:
 464             return
 465
 466         # If we have a context, then use the context to find any
 467         # request waiting for a reply
 468         context = 0
 469         if hasattr(r, 'context') and r.context > 0:
 470             context = r.context
 471
 472         msgname = type(r).__name__
 473
 474         if context == 0:
 475             # No context -> async notification that we feed to the callback
 476             if self.event_callback:
 477                 self.event_callback(msgname, r)
 478         else:
 479             # Context -> use the results structure (carefully) to find
 480             # who we're responding to and return the message to that
 481             # thread
 482             with self.results_lock:
 483                 if context not in self.results:
 484                     eprint('Not expecting results for this context', context, r)
 485                 else:
 486                     result = self.results[context]
 487
 488                     #
 489                     # Collect results until control ping
 490                     #
 491
 492                     if msgname == 'control_ping_reply':
 493                         # End of a multipart
 494                         result['e'].set()
 495                     elif 'm' in self.results[context]:
 496                         # One element in a multipart
 497                         result['r'].append(r)
 498                     else:
 499                         # All of a single result
 500                         result['r'] = r
 501                         result['e'].set()
 502
 503     def decode_incoming_msg(self, msg):
 504         if not msg:
 505             eprint('vpp_api.read failed')
 506             return
 507
 508         i, ci = self.header.unpack_from(msg, 0)
 509         if self.id_names[i] == 'rx_thread_exit':
 510             return
 511
 512         #
 513         # Decode message and returns a tuple.
 514         #
 515         msgdef = self.id_msgdef[i]
 516         if not msgdef:
 517             raise IOError(2, 'Reply message undefined')
 518
 519         r = self.decode(msgdef, msg)
 520
 521         return r
 522
 523     def msg_handler_async(self, msg):
 524         """Process a message from VPP in async mode.
 525
 526         In async mode, all messages are returned to the callback.
 527         """
 528         r = self.decode_incoming_msg(msg)
 529         if r is None:
 530             return
 531
 532         msgname = type(r).__name__
 533
 534         if self.event_callback:
 535             self.event_callback(msgname, r)
 536
 537     def _control_ping(self, context):
 538         """Send a ping command."""
 539         self._call_vpp_async(self.control_ping_index,
 540                              self.control_ping_msgdef,
 541                              context=context)
 542
 543     def _call_vpp(self, i, msgdef, multipart, **kwargs):
 544         """Given a message, send the message and await a reply.
 545
 546         msgdef - the message packing definition
 547         i - the message type index
 548         multipart - True if the message returns multiple
 549         messages in return.
 550         context - context number - chosen at random if not
 551         supplied.
 552         The remainder of the kwargs are the arguments to the API call.
 553
 554         The return value is the message or message array containing
 555         the response.  It will raise an IOError exception if there was
 556         no response within the timeout window.
 557         """
 558
 559         # We need a context if not supplied, in order to get the
 560         # response
 561         context = kwargs.get('context', self.get_context())
 562         kwargs['context'] = context
 563
 564         # Set up to receive a response
 565         self.results_prepare(context, multi=multipart)
 566
 567         # Output the message
 568         self._call_vpp_async(i, msgdef, **kwargs)
 569
 570         if multipart:
 571             # Send a ping after the request - we use its response
 572             # to detect that we have seen all results.
 573             self._control_ping(context)
 574
 575         # Block until we get a reply.
 576         r = self.results_wait(context)
 577
 578         return r
 579
 580     def _call_vpp_async(self, i, msgdef, **kwargs):
 581         """Given a message, send the message and await a reply.
 582
 583         msgdef - the message packing definition
 584         i - the message type index
 585         context - context number - chosen at random if not
 586         supplied.
 587         The remainder of the kwargs are the arguments to the API call.
 588         """
 589         if not 'context' in kwargs:
 590             context = self.get_context()
 591             kwargs['context'] = context
 592         else:
 593             context = kwargs['context']
 594         kwargs['_vl_msg_id'] = i
 595         b = self.encode(msgdef, kwargs)
 596
 597         self._write(b)
 598
 599     def register_event_callback(self, callback):
 600         """Register a callback for async messages.
 601
 602         This will be called for async notifications in sync mode,
 603         and all messages in async mode.  In sync mode, replies to
 604         requests will not come here.
 605
 606         callback is a fn(msg_type_name, msg_type) that will be
 607         called when a message comes in.  While this function is
 608         executing, note that (a) you are in a background thread and
 609         may wish to use threading.Lock to protect your datastructures,
 610         and (b) message processing from VPP will stop (so if you take
 611         a long while about it you may provoke reply timeouts or cause
 612         VPP to fill the RX buffer).  Passing None will disable the
 613         callback.
 614         """
 615         self.event_callback = callback