I have written a thread-safe list which can be used as a drop-in replacement for a regular list. This has been tested and used internally without issue, but "The absence of evidence is not evidence of absence" so even though I have had no issues with using this, do you folks see anything wrong with the code?
Stepping back, here is the original problem and why threading was introduced to help solve it.
In Cisco's ACI solution, the APIC is a central point of management for the network. It features a declarative configuration format which can be queried and updated via it's REST API.
One interesting feature is the ability for clients to subscribe to a query (scroll up slightly) and receive push notifications via a websocket whenever the result of that query changes.
I have written a class to simplify managing connection state on behalf of clients who are querying the APIC. It is used like this:
from MyModule import MySessionClass
session = MySessionClass(my_apic_ip)
session.post('/uni/<some-managed-object-path-here>', payload)
session.subscribe('/uni/<watch_this_mo>') # Sends GET request to APIC, appends to RLockedList
session.subscribe('/uni/<watch_this_mo_2>')
session.subscribe('/uni/<watch_this_mo_3>')
...
session.unsubscribe('/uni/<watch_this_mo_2>') # Removes this dn from the RLockedList, after 60s or less we stop receiving push notifications for this query on the websocket.
...
session.unsubscribe('/uni/<watch_this_mo>')
session.unsubscribe('/uni/<watch_this_mo_3>')
Due to the following requirements, I have introduced threading:
- We must refresh our API session (via POST) before 5 minutes have passed (in the future, this may have multiple API sessions to refresh due to controller clustering.)
- We must refresh each push-notification subscription we have every minute (or less)
- We must listen for push notifications and receive them in a timely manner
The above is complicated by the following:
- Excessively large read and update (GET and POST) operations can take minutes to return in some cases. In other cases (e.g the result is still cached on the APIC) the same query will return quickly. Without threads the subscriptions would certainly time out in this case.
- Multiple subscriptions to the same session can be active at one time.
- While one thread is refreshing the subscriptions in the list, another thread may modify the list. It can remove and delete an element from anywhere in the list - usually somewhere in the middle, rather than at the ends.
This thread-safe list will maintain the current list of active subscriptions. The subscription refresh thread will periodically iterate through this list, sending GET requests to the APIC. These generally return in ~1 second and we expect the maximum number of user subscriptions to be 30, though there is no hard restriction on this on the server.
So there is one thread for refreshing the subscriptions, one thread to keep us logged in, and one thread to read the push notifications on the websocket as they come in.
This is the list used to manage subscriptions. Does this implementation of a re-entrant locked list look thread-safe to you? By this, I mean truly thread safe, as in thread-safe within an environment with multiple kernel threads (as is seen in Jython, for example) and without the assurance of the Global Interpreter Lock. Most likely, this is fine. But if anyone is an expert in threading in Python, I would appreciate it if you looked this over.
import threading
class RLockedList(list):
def __init__(self, in_list = []):
self.__in_list = in_list # Internal List
self.__lock = threading.RLock()
def __add__(self, x):
with self.__lock:
return list.__add__(self.__in_list, x)
def __contains__(self, item):
with self.__lock:
return list.__contains__(self.__in_list, item)
def __class__(self):
return RLockedList
def __delitem__(self, key):
with self.__lock:
return list.__delitem__(self.__in_list, key)
def __delslice__(self, i, j):
with self.__lock:
return list.__delslice__(self.__in_list, i, j)
def __doc__(self):
return list.__doc__(self.__in_list, i, j)
def __eq__(self, other):
with self.__lock:
return list.__eq__(self.__in_list, other)
def __format__(self, format_spec):
with self.__lock:
return list.__format__(self.__in_list, format_spec)
def __ge__(self, other):
with self.__lock:
return list.__ge__(self.__in_list, other)
def __getitem__(self, i):
with self.__lock:
return list.__getitem__(self.__in_list, i)
def __getslice__(self, i, j):
with self.__lock:
return list.__getslice__(self.__in_list, i, j)
def __gt__(self, other):
with self.__lock:
return list.__gt__(self.__in_list, other)
def __hash__(self):
with self.__lock:
return list.__hash__(self.__in_list, other)
def __iadd__(self, other):
with self.__lock:
return list.__iadd__(self.__in_list, other)
def __imul__(self, other):
with self.__lock:
return list.__imul__(self.__in_list, other)
def __iter__(self):
'''return an iterator to a copy of the list, so that thread-safety is maintained.'''
with self.__lock:
return list.__iter__(list(self.__in_list))
def __le__(self, other):
with self.__lock:
return list.__le__(self.__in_list, other)
def __len__(self):
with self.__lock:
return list.__len__(self.__in_list)
def __lt__(self, other):
with self.__lock:
return list.__lt__(self.__in_list, other)
def __mul__(self, other):
with self.__lock:
return list.__mul__(self.__in_list, other)
def __ne__(self, other):
with self.__lock:
return list.__ne__(self.__in_list, other)
def __reduce__(self):
with self.__lock:
return list.__reduce__(self.__in_list)
def __reduce_ex__(self):
with self.__lock:
return list.__reduce_ex__(self.__in_list)
def __repr__(self):
with self.__lock:
return list.__repr__(self.__in_list)
def __reversed__(self):
with self.__lock:
return list.__reversed__(self.__in_list)
def __rmul__(self, other):
with self.__lock:
return list.__rmul__(self.__in_list, other)
def __setitem__(self, key, value):
with self.__lock:
return list.__setitem__(self.__in_list, key, value)
def __setslice__(self, i, j, sequence):
with self.__lock:
return list.__setslice__(self.__in_list, i, j, sequence)
def __sizeof__(self):
with self.__lock:
return list.__sizeof__(self.__in_list)
def __str__(self):
with self.__lock:
return list.__str__(self.__in_list)
def __subclasshook__(self, subclass):
with self.__lock:
return list.__subclasshook__(self.in_list, subclass)
def append(self, x):
with self.__lock:
return list.append(self.__in_list, x)
def count(self, x):
with self.__lock:
return list.count(self.__in_list, x)
def extend(self, L):
with self.__lock:
return list.extend(self.__in_list, L)
def index(self, x):
with self.__lock:
return list.index(self.__in_list, x)
def insert(self, i, x):
with self.__lock:
return list.insert(self.__in_list, i, x)
def pop(self, i = -1):
with self.__lock:
return list.pop(self.__in_list, i)
def remove(self, x):
with self.__lock:
return list.remove(self.__in_list, x)
def reverse(self):
with self.__lock:
return list.reverse(self.__in_list)
def sort(self, _cmp=None, key=None, reverse=False):
with self.__lock:
return list.sort(self.__in_list, _cmp, key, reverse)
def get_list(self):
'''return a copy of the current state of the thread-safe list.'''
with self.__lock:
return list(self.__in_list)
collections.dequeorqueue? \$\endgroup\$