vpp_transport_socket: make connect more resilient
This should make connect() partially retriable,
at least against "No such file or directory on socket",
and when disconnect() is called before retrying connect().
Added TODOs for future improvements.
@@ -11,6 +11,7 @@ try:
except ImportError:
import Queue as queue
import logging
+from . import vpp_papi
class VppTransportSocketIOError(IOError):
@@ -28,9 +29,11 @@ class VppTransport(object):
self.server_address = server_address
self.header = struct.Struct('>QII')
self.message_table = {}
- self.sque = multiprocessing.Queue()
- self.q = multiprocessing.Queue()
- self.message_thread = None # Will be set on connect().
+ # The following fields are set in connect().
+ self.sque = None
+ self.q = None
+ self.message_thread = None
+ self.socket = None
def msg_thread_func(self):
while True:
@@ -69,11 +72,12 @@ class VppTransport(object):
2, 'Unknown response from select')
def connect(self, name, pfx, msg_handler, rx_qlen):
+ # TODO: Reorder the actions and add "roll-backs",
+ # to restore clean disconnect state when failure happens durng connect.
if self.message_thread is not None:
raise VppTransportSocketIOError(
1, "PAPI socket transport connect: Need to disconnect first.")
- self.message_thread = threading.Thread(target=self.msg_thread_func)
# Create a UDS socket
self.socket = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
@@ -87,6 +91,12 @@ class VppTransport(object):
raise
self.connected = True
+
+ # TODO: Can this block be moved even later?
+ self.sque = multiprocessing.Queue()
+ self.q = multiprocessing.Queue()
+ self.message_thread = threading.Thread(target=self.msg_thread_func)
+
# Initialise sockclnt_create
sockclnt_create = self.parent.messages['sockclnt_create']
sockclnt_create_reply = self.parent.messages['sockclnt_create_reply']
@@ -114,18 +124,29 @@ class VppTransport(object):
return 0
def disconnect(self):
- # TODO: Should we detect if user forgot to connect first?
+ # TODO: Support repeated disconnect calls, recommend users to call
+ # disconnect when they are not sure what the state is after failures.
+ # TODO: Any volunteer for comprehensive docstrings?
rv = 0
- try: # Might fail, if VPP closes socket before packet makes it out
+ try:
+ # Might fail, if VPP closes socket before packet makes it out,
+ # or if there was a failure during connect().
rv = self.parent.api.sockclnt_delete(index=self.socket_index)
- except IOError:
+ except (IOError, vpp_papi.VPPApiError):
pass
self.connected = False
+ if self.socket is not None:
self.socket.close()
+ if self.sque is not None:
self.sque.put(True) # Terminate listening thread
- self.message_thread.join()
+ if self.message_thread is not None:
# Allow additional connect() calls.
+ self.message_thread.join()
+ # Collect garbage.
+ self.sque = None
+ self.q = None
self.message_thread = None
+ self.socket = None
return rv
def suspend(self):
