Merge "Mutex access to local lock attributes"

nodepool/zk/zookeeper.py

@@ -15,6 +15,7 @@ from copy import copy
 import abc
 import json
 import logging
+import threading
 import time
 import uuid
 
@@ -394,6 +395,9 @@ class NodeRequest(BaseModel):
     def __init__(self, id=None):
         super(NodeRequest, self).__init__(id)
         self.lock = None
+        # Local thread lock that is acquired when we are manipulating
+        # the ZK lock.
+        self._thread_lock = threading.Lock()
         self.declined_by = []
         self.node_types = []
         self.nodes = []
@@ -510,6 +514,9 @@ class Node(BaseModel):
         super(Node, self).__init__(id)
         # Local lock object; not serialized
         self.lock = None
+        # Local thread lock that is acquired when we are manipulating
+        # the ZK lock.
+        self._thread_lock = threading.Lock()
         # Cached list of lock contenders; not serialized (and possibly
         # not up to date; use for status listings only).
         self.lock_contenders = set()
@@ -2138,6 +2145,7 @@ class ZooKeeper(ZooKeeperBase):
         log = get_annotated_logger(self.log, event_id=request.event_id,
                                    node_request_id=request.id)
         path = self._requestLockPath(request.id)
+        with request._thread_lock:
             try:
                 lock = Lock(self.kazoo_client, path)
                 have_lock = lock.acquire(blocking, timeout)
@@ -2171,6 +2179,7 @@ class ZooKeeper(ZooKeeperBase):
         if request.lock is None:
             raise npe.ZKLockException(
                 "Request %s does not hold a lock" % request)
+        with request._thread_lock:
             request.lock.release()
             request.lock = None
 
@@ -2199,6 +2208,7 @@ class ZooKeeper(ZooKeeperBase):
             and could not get the lock, or a lock is already held.
         '''
         path = self._nodeLockPath(node.id)
+        with node._thread_lock:
             try:
                 lock = Lock(self.kazoo_client, path, identifier)
                 have_lock = lock.acquire(blocking, timeout, ephemeral)
@@ -2231,6 +2241,7 @@ class ZooKeeper(ZooKeeperBase):
         '''
         if node.lock is None:
             raise npe.ZKLockException("Node %s does not hold a lock" % node)
+        with node._thread_lock:
             node.lock.release()
             node.lock = None