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

   1 #
   2 # VPP Unix Domain Socket Transport.
   3 #
   4 import socket
   5 import struct
   6 import threading
   7 import select
   8 import multiprocessing
   9 import queue
  10 import logging
  11
  12 logger = logging.getLogger("vpp_papi.transport")
  13 logger.addHandler(logging.NullHandler())
  14
  15
  16 class VppTransportSocketIOError(IOError):
  17     # TODO: Document different values of error number (first numeric argument).
  18     pass
  19
  20
  21 class VppTransport:
  22     VppTransportSocketIOError = VppTransportSocketIOError
  23
  24     def __init__(self, parent, read_timeout, server_address):
  25         self.connected = False
  26         self.read_timeout = read_timeout if read_timeout > 0 else None
  27         self.parent = parent
  28         self.server_address = server_address
  29         self.header = struct.Struct(">QII")
  30         self.message_table = {}
  31         # These queues can be accessed async.
  32         # They are always up, but replaced on connect.
  33         # TODO: Use multiprocessing.Pipe instead of multiprocessing.Queue
  34         # if possible.
  35         self.sque = multiprocessing.Queue()
  36         self.q = multiprocessing.Queue()
  37         # The following fields are set in connect().
  38         self.message_thread = None
  39         self.socket = None
  40
  41     def msg_thread_func(self):
  42         while True:
  43             try:
  44                 rlist, _, _ = select.select([self.socket, self.sque._reader], [], [])
  45             except socket.error:
  46                 # Terminate thread
  47                 logging.error("select failed")
  48                 self.q.put(None)
  49                 return
  50
  51             for r in rlist:
  52                 if r == self.sque._reader:
  53                     # Terminate
  54                     self.q.put(None)
  55                     return
  56
  57                 elif r == self.socket:
  58                     try:
  59                         msg = self._read()
  60                         if not msg:
  61                             self.q.put(None)
  62                             return
  63                     except socket.error:
  64                         self.q.put(None)
  65                         return
  66                     # Put either to local queue or if context == 0
  67                     # callback queue
  68                     if self.parent.has_context(msg):
  69                         self.q.put(msg)
  70                     else:
  71                         self.parent.msg_handler_async(msg)
  72                 else:
  73                     raise VppTransportSocketIOError(2, "Unknown response from select")
  74
  75     def connect(self, name, pfx, msg_handler, rx_qlen):
  76         # TODO: Reorder the actions and add "roll-backs",
  77         # to restore clean disconnect state when failure happens durng connect.
  78
  79         if self.message_thread is not None:
  80             raise VppTransportSocketIOError(
  81                 1, "PAPI socket transport connect: Need to disconnect first."
  82             )
  83
  84         # Create a UDS socket
  85         self.socket = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
  86         self.socket.settimeout(self.read_timeout)
  87
  88         # Connect the socket to the port where the server is listening
  89         try:
  90             self.socket.connect(self.server_address)
  91         except socket.error as msg:
  92             # logging.error("{} on socket {}".format(msg, self.server_address))
  93             raise msg
  94
  95         self.connected = True
  96
  97         # Queues' feeder threads from previous connect may still be sending.
  98         # Close and join to avoid any errors.
  99         self.sque.close()
 100         self.q.close()
 101         self.sque.join_thread()
 102         self.q.join_thread()
 103         # Finally safe to replace.
 104         self.sque = multiprocessing.Queue()
 105         self.q = multiprocessing.Queue()
 106         self.message_thread = threading.Thread(target=self.msg_thread_func)
 107
 108         # Initialise sockclnt_create
 109         sockclnt_create = self.parent.messages["sockclnt_create"]
 110         sockclnt_create_reply = self.parent.messages["sockclnt_create_reply"]
 111
 112         args = {"_vl_msg_id": 15, "name": name, "context": 124}
 113         b = sockclnt_create.pack(args)
 114         self.write(b)
 115         msg = self._read()
 116         hdr, length = self.parent.header.unpack(msg, 0)
 117         if hdr.msgid != 16:
 118             # TODO: Add first numeric argument.
 119             raise VppTransportSocketIOError("Invalid reply message")
 120
 121         r, length = sockclnt_create_reply.unpack(msg)
 122         self.socket_index = r.index
 123         for m in r.message_table:
 124             n = m.name
 125             self.message_table[n] = m.index
 126
 127         self.message_thread.daemon = True
 128         self.message_thread.start()
 129
 130         return 0
 131
 132     def disconnect(self):
 133         # TODO: Support repeated disconnect calls, recommend users to call
 134         # disconnect when they are not sure what the state is after failures.
 135         # TODO: Any volunteer for comprehensive docstrings?
 136         rv = 0
 137         try:
 138             # Might fail, if VPP closes socket before packet makes it out,
 139             # or if there was a failure during connect().
 140             # TODO: manually build message so that .disconnect releases server-side resources
 141             rv = self.parent.api.sockclnt_delete(index=self.socket_index)
 142         except (IOError, self.parent.VPPApiError):
 143             pass
 144         self.connected = False
 145         if self.socket is not None:
 146             self.socket.close()
 147         if self.sque is not None:
 148             self.sque.put(True)  # Terminate listening thread
 149         if self.message_thread is not None and self.message_thread.is_alive():
 150             # Allow additional connect() calls.
 151             self.message_thread.join()
 152         # Wipe message table, VPP can be restarted with different plugins.
 153         self.message_table = {}
 154         # Collect garbage.
 155         self.message_thread = None
 156         self.socket = None
 157         # Queues will be collected after connect replaces them.
 158         return rv
 159
 160     def suspend(self):
 161         pass
 162
 163     def resume(self):
 164         pass
 165
 166     def callback(self):
 167         raise NotImplementedError
 168
 169     def get_callback(self, do_async):
 170         return self.callback
 171
 172     def get_msg_index(self, name):
 173         try:
 174             return self.message_table[name]
 175         except KeyError:
 176             return 0
 177
 178     def msg_table_max_index(self):
 179         return len(self.message_table)
 180
 181     def write(self, buf):
 182         """Send a binary-packed message to VPP."""
 183         if not self.connected:
 184             raise VppTransportSocketIOError(1, "Not connected")
 185
 186         # Send header
 187         header = self.header.pack(0, len(buf), 0)
 188         try:
 189             self.socket.sendall(header)
 190             self.socket.sendall(buf)
 191         except socket.error as err:
 192             raise VppTransportSocketIOError(1, "Sendall error: {err!r}".format(err=err))
 193
 194     def _read_fixed(self, size):
 195         """Repeat receive until fixed size is read. Return empty on error."""
 196         buf = bytearray(size)
 197         view = memoryview(buf)
 198         left = size
 199         while 1:
 200             got = self.socket.recv_into(view, left)
 201             if got <= 0:
 202                 # Read error.
 203                 return ""
 204             if got >= left:
 205                 # TODO: Raise if got > left?
 206                 break
 207             left -= got
 208             view = view[got:]
 209         return buf
 210
 211     def _read(self):
 212         """Read single complete message, return it or empty on error."""
 213         hdr = self._read_fixed(16)
 214         if not hdr:
 215             return
 216         (_, hdrlen, _) = self.header.unpack(hdr)  # If at head of message
 217
 218         # Read rest of message
 219         msg = self._read_fixed(hdrlen)
 220         if hdrlen == len(msg):
 221             return msg
 222         raise VppTransportSocketIOError(1, "Unknown socket read error")
 223
 224     def read(self, timeout=None):
 225         if not self.connected:
 226             raise VppTransportSocketIOError(1, "Not connected")
 227         if timeout is None:
 228             timeout = self.read_timeout
 229         try:
 230             return self.q.get(True, timeout)
 231         except queue.Empty:
 232             return None