CP-39600: Rework LVM locking to use fair lock queue (xapi-project#658)

This is intended to prevent a storm of later LVM commands from usurping
the lock that an earlier command is waiting for, just because they
happened to manage to start waiting before the original holder was done.

In some pathalogical circumstances, this has been known to result in a
commant taking many minutes to get the lock.

Signed-off-by: Tim Smith <tim.smith@cloud.com>
Co-authored-by: Tim Smith <tim.smith@cloud.com>

Makefile

@@ -45,6 +45,7 @@ SM_LIBS += journaler
 SM_LIBS += fjournaler
 SM_LIBS += lock
 SM_LIBS += flock
+SM_LIBS += lock_queue
 SM_LIBS += ipc
 SM_LIBS += srmetadata
 SM_LIBS += metadata

drivers/LVHDSR.py

@@ -1758,9 +1758,7 @@ def _snapshot(self, snapType, cloneOp=False, cbtlog=None, cbt_consistency=None):
         if snapType == VDI.SNAPSHOT_DOUBLE:
             clonUuid = util.gen_uuid()
         jval = "%s_%s" % (baseUuid, clonUuid)
-        with lvutil.LvmLockContext():
-            # This makes multiple LVM calls so take the lock early
-            self.sr.journaler.create(self.JRN_CLONE, origUuid, jval)
+        self.sr.journaler.create(self.JRN_CLONE, origUuid, jval)
         util.fistpoint.activate("LVHDRT_clone_vdi_after_create_journal", self.sr.uuid)
 
         try:
@@ -1835,9 +1833,7 @@ def _snapshot(self, snapType, cloneOp=False, cbtlog=None, cbt_consistency=None):
             self._failClone(origUuid, jval, str(e))
         util.fistpoint.activate("LVHDRT_clone_vdi_before_remove_journal", self.sr.uuid)
 
-        with lvutil.LvmLockContext():
-            # This makes multiple LVM calls so take the lock early
-            self.sr.journaler.remove(self.JRN_CLONE, origUuid)
+        self.sr.journaler.remove(self.JRN_CLONE, origUuid)
 
         return self._finishSnapshot(snapVDI, snapVDI2, hostRefs, cloneOp, snapType)
 

drivers/lock.py

@@ -49,7 +49,6 @@ def __new__(cls, name, ns=None, *args, **kwargs):
             instances[name] = LockImplementation(name, ns)
         return instances[name]
 
-    # These are required to pacify pylint as it doesn't understand the __new__
     def acquire(self):
         raise NotImplementedError("Lock methods implemented in LockImplementation")
 

drivers/lock_queue.py

@@ -0,0 +1,120 @@
+# Copyright (C) Cloud Software Group, Inc.
+#
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU Lesser General Public License as published
+# by the Free Software Foundation; version 2.1 only.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU Lesser General Public License for more details.
+#
+# You should have received a copy of the GNU Lesser General Public License
+# along with this program; if not, write to the Free Software Foundation, Inc.,
+# 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301  USA
+
+import os
+import pickle
+import sys
+import time
+
+import lock
+import util
+
+
+DEBUG_LOG = True
+
+def debug_log(msg):
+    if DEBUG_LOG:
+        util.SMlog("LockQueue: " + msg)
+
+def get_process_start_time(pid):
+    proc_file = f"/proc/{pid}/stat"
+    with open(proc_file, 'r') as f:
+        return f.read().split(')')[-1].split(' ')[20]
+
+def process_is_valid(pid, start_time):
+    proc_file = f"/proc/{pid}/stat"
+
+    try:
+        if start_time != get_process_start_time(pid):
+            debug_log(f"Process {pid} has incorrect start time real:{get_process_start_time(pid)} vs expected:{start_time}")
+            return False
+    except FileNotFoundError:
+        debug_log(f"Process {pid} is dead")
+        return False
+
+    return True
+
+class LockQueue:
+    def __init__(self, name):
+        self.name = name
+        self._queue_lock = lock.Lock(name, f"ql-{name}")
+        self._action_lock = lock.Lock(name, f"al-{name}")
+        # Filename to hold the process queue
+        self._mem = f"/tmp/mem-{name}"
+
+    def load_queue(self):
+        try:
+            with open(self._mem, "rb") as f:
+                queue = pickle.load(f)
+            debug_log("load_queue {}".format(queue))
+        except EOFError:
+            queue = []
+        except FileNotFoundError:
+            queue = []
+        return queue
+
+    def save_queue(self, queue):
+        with open(self._mem, "w+b") as f:
+            pickle.dump(queue, f)
+        debug_log("save_queue {}".format(queue))
+
+    def push_into_process_queue(self):
+        self._queue_lock.acquire()
+
+        queue = self.load_queue()
+        queue.append((os.getpid(), get_process_start_time(os.getpid())))
+        self.save_queue(queue)
+
+        self._queue_lock.release()
+
+    def __enter__(self):
+        # Add ourselves to the process queue.
+        self.push_into_process_queue()
+
+        # Keep reading the process queue until we are at the front
+        while True:
+            self._queue_lock.acquire()
+            queue = self.load_queue()
+            front_pid, front_start_time = queue.pop(0)
+            print(f"Testing for PID {front_pid}")
+            if front_pid == os.getpid():
+                # We are at the front, it is now our turn to wait on the action lock
+                # and then do our work
+                debug_log(f"{front_pid} taking action lock")
+                self._action_lock.acquire()
+                # When we have the action lock, save the queue (which no longer
+                # includes us) and release the queue lock to let others join.
+                self.save_queue(queue)
+                self._queue_lock.release()
+                break
+
+            # Getting here means it was not our turn to do stuff
+            # If the process at the front of the queue is not alive then remove it
+            if not process_is_valid(front_pid, front_start_time):
+                # front pid has already been popped from queue so just save it
+                debug_log(f"Removing invalid process {front_pid}")
+                self.save_queue(queue)
+            # Release the lock and try again later. Most waiting will be on the queue lock,
+            # waiting for the single Action lock waiter to release it when it has the action
+            # lock. We sleep a short while before our next check to make it easier for new
+            # waiters to join the queue without really wasting our own time.
+            self._queue_lock.release()
+            time.sleep(0.1)
+
+        debug_log("In manager")
+        return self
+
+    def __exit__(self, type, value, tbck):
+        self._action_lock.release()

drivers/lvutil.py

@@ -16,12 +16,14 @@
 # Miscellaneous LVM utility functions
 #
 
+import traceback
 import re
 import os
 import errno
 import time
 
 import lock
+import lock_queue
 import util
 import xs_errors
 import xml.dom.minidom
@@ -110,7 +112,7 @@ def extract_vgname(str_in):
 
     return None
 
-class LvmLockContext(object):
+class LvmLockContext(lock_queue.LockQueue):
     """
     Context manager around the LVM lock.
 
@@ -120,20 +122,35 @@ class LvmLockContext(object):
     """
 
     def __init__(self, cmd=None):
-        self.lock = lock.Lock(LVM_LOCK)
         self.cmd = cmd
         self.locked = False
+        try:
+            super().__init__(LVM_LOCK)
+        except Exception as e:
+            util.SMlog(f"===> LvmLockContext __init__ {e} {traceback.format_exc()}")
+            raise
 
     def __enter__(self):
         if self.cmd and '--readonly' in self.cmd:
             return
 
-        self.lock.acquire()
+        while True:
+            try:
+                super().__enter__()
+            except Exception as e:
+                util.SMlog(f"===> LvmLockContext __enter__ {e} {traceback.format_exc()}")
+                continue
+            break
+
         self.locked = True
 
-    def __exit__(self, exc_type, value, traceback):
+    def __exit__(self, exc_type, value, tbck):
         if self.locked:
-            self.lock.release()
+            try:
+                super().__exit__(exc_type, value, tbck)
+            except Exception as e:
+                util.SMlog(f"===> LvmLockContext __exit__ {e} {traceback.format_exc()}")
+                raise
 
 
 LVM_RETRY_ERRORS = [

mk/sm.spec.in

@@ -169,6 +169,7 @@ tests/run_python_unittests.sh
 /opt/xensource/sm/journaler.py
 /opt/xensource/sm/lcache.py
 /opt/xensource/sm/lock.py
+/opt/xensource/sm/lock_queue.py
 /opt/xensource/sm/lvhdutil.py
 /opt/xensource/sm/lvmanager.py
 /opt/xensource/sm/lvmcache.py

tests/test_LVHDSR.py

@@ -54,10 +54,11 @@ def create_LVHDSR(self, master=False, command='foo', sr_uuid=None):
             sr_uuid = str(uuid.uuid4())
         return LVHDSR.LVHDSR(srcmd, sr_uuid)
 
+    @mock.patch('lvutil.LvmLockContext', autospec=True)
     @mock.patch('lvhdutil.getVDIInfo', autospec=True)
     @mock.patch('LVHDSR.Lock', autospec=True)
     @mock.patch('SR.XenAPI')
-    def test_loadvids(self, mock_xenapi, mock_lock, mock_getVDIInfo):
+    def test_loadvids(self, mock_xenapi, mock_lock, mock_getVDIInfo, mock_lvlock):
         """sr.allVDIs populated by _loadvdis"""
 
         vdi_uuid = 'some VDI UUID'

tests/test_LVHDoHBASR.py

@@ -87,6 +87,8 @@ def setUp(self):
         self.mock_lvhdsr = lvhdsr_patcher.start()
         util_patcher = mock.patch('LVHDoHBASR.util', autospec=True)
         self.mock_util = util_patcher.start()
+        lc_patcher = mock.patch('LVHDSR.lvmcache.lvutil.LvmLockContext', autospec=True)
+        self.mock_lc = lc_patcher.start()
         xenapi_patcher = mock.patch('SR.XenAPI')
         self.mock_xapi = xenapi_patcher.start()
 

tests/test_LVHDoISCSISR.py

@@ -148,6 +148,8 @@ def deepcopy(to_copy):
 
         lock_patcher = mock.patch('LVHDSR.Lock')
         self.mock_lock = lock_patcher.start()
+        lvlock_patcher = mock.patch('LVHDSR.lvutil.LvmLockContext')
+        self.mock_lvlock = lvlock_patcher.start()
 
         self.addCleanup(mock.patch.stopall)
 

tests/test_SRCommand.py

@@ -126,11 +126,13 @@ def test_run_wrapped_if_not_SRException(
                 new=lambda s: s.encode("ascii", "surrogateescape"))
     @mock.patch("os.fsdecode",
                 new=lambda bs: bs.decode("ascii", "surrogateescape"))
-    def test_parse_handles_wide_chars(self):
+    @mock.patch('util.gen_uuid', autospec=True)
+    def test_parse_handles_wide_chars(self, gen_uuid):
         import os
         import xmlrpc.client
         from DummySR import DRIVER_INFO
 
+        gen_uuid.return_value = '13c4384e-897b-e745-6b3e-9a89c06537be'
         xmlrpc_method = "vdi_create"
         xmlrpc_params = {
             'host_ref': 'OpaqueRef:133c7c46-f4d9-3695-83c4-bf8574b89fb9',

tests/test_lock_queue.py

@@ -0,0 +1,87 @@
+import builtins
+import copy
+import os
+import sys
+import unittest
+import unittest.mock as mock
+
+import lock
+import lock_queue
+
+## Instead of saving the process queue to disk the mocks will save it here.
+## It needs to be global because it is shared between threads.
+saved_queue = []
+
+def mock_pickle_dump_fn(*args):
+   global saved_queue
+   saved_queue = copy.deepcopy(args[0])
+
+def mock_pickle_load_fn(*args):
+   global saved_queue
+   return copy.deepcopy(saved_queue)
+
+
+class Test_LockQueue(unittest.TestCase):
+    def setUp(self):
+        # Re-initialize queue to empty for each test
+        global saved_queue
+        saved_queue = []
+
+    def get_lock_name(self):
+        return "bacon"
+
+    @mock.patch('lock_queue.pickle.load', side_effect=mock_pickle_load_fn)
+    @mock.patch('lock_queue.pickle.dump', side_effect=mock_pickle_dump_fn)
+    @mock.patch('lock_queue.os.getpid')
+    @mock.patch('lock_queue.get_process_start_time')
+    @mock.patch('lock.Lock', autospec=False)
+    def test_push_to_queue_3x(self, lock, start_time, getpid, pdump, pload):
+        global saved_queue
+
+        lq = lock_queue.LockQueue(self.get_lock_name())
+        # Push to queue 3 times using these PID and Start Time combinations
+        test_pids = [997, 993, 996]
+        test_sts = [360, 430, 458]
+        for p, s in zip(test_pids, test_sts):
+            start_time.return_value = s
+            getpid.return_value = p
+            lq.push_into_process_queue()
+
+        # Test the queue includes the PID and Start Time pairs in the order we expect
+        self.assertEqual(list(zip(test_pids, test_sts)), saved_queue)
+
+    @mock.patch('lock_queue.pickle.load', side_effect=mock_pickle_load_fn)
+    @mock.patch('lock_queue.pickle.dump', side_effect=mock_pickle_dump_fn)
+    @mock.patch('lock_queue.os.getpid')
+    @mock.patch('lock_queue.get_process_start_time')
+    @mock.patch('lock.Lock', autospec=False)
+    def test_context_manager(self, lock, start_time, getpid, pdump, pload):
+        global saved_queue
+
+        getpid.return_value = 959
+        start_time.return_value = 575
+
+        # Queue is empty
+        self.assertEqual(saved_queue, [])
+
+        with lock_queue.LockQueue(self.get_lock_name()) as lq:
+            # Should have removed from the queue before completing entry to the context manager
+            self.assertEqual(saved_queue, [])
+
+    @mock.patch('lock_queue.pickle.load', side_effect=mock_pickle_load_fn)
+    @mock.patch('lock_queue.pickle.dump', side_effect=mock_pickle_dump_fn)
+    @mock.patch('lock_queue.os.getpid')
+    @mock.patch('lock_queue.get_process_start_time')
+    @mock.patch('lock.Lock', autospec=False)
+    def test_context_manager_bad_entry(self, lock, start_time, getpid, pdump, pload):
+        global saved_queue
+
+        # Initialise saved_queue with non-existent pid
+        saved_queue = [(0, 67867)]
+
+        getpid.return_value = 959
+        start_time.return_value = 575
+        with lock_queue.LockQueue(self.get_lock_name()) as lq:
+            # Should have removed from the queue before completing entry to the context manager
+            self.assertEqual(saved_queue, [])
+