| # -*- coding: utf-8 -*- |
| # Copyright 2015 The Chromium OS Authors. All rights reserved. |
| # Use of this source code is governed by a BSD-style license that can be |
| # found in the LICENSE file. |
| |
| """Test the locking library.""" |
| |
| from __future__ import print_function |
| |
| import itertools |
| import multiprocessing |
| import os |
| import sys |
| import time |
| |
| from chromite.lib import cros_test_lib |
| from chromite.lib import locking |
| from chromite.lib import parallel |
| from chromite.lib import osutils |
| from chromite.lib import process_util |
| from chromite.lib import timeout_util |
| |
| |
| LOCK_ACQUIRED = 5 |
| LOCK_NOT_ACQUIRED = 6 |
| |
| |
| class LockingTest(cros_test_lib.TempDirTestCase): |
| """Test the Locking class.""" |
| |
| def setUp(self): |
| self.lock_file = os.path.join(self.tempdir, 'lockfile') |
| |
| def _HelperSingleLockTest(self, blocking, shared, locktype): |
| """Helper method that runs a basic test with/without blocking/sharing.""" |
| self.assertNotExists(self.lock_file) |
| |
| lock = locking.FileLock( |
| self.lock_file, blocking=blocking, locktype=locktype) |
| self.assertFalse(lock.IsLocked()) |
| |
| lock.write_lock() |
| self.assertTrue(lock.IsLocked()) |
| self.assertExists(self.lock_file) |
| |
| # Acquiring the lock again should be safe. |
| lock.lock(shared) |
| self.assertTrue(lock.IsLocked()) |
| |
| lock.close() |
| self.assertFalse(lock.IsLocked()) |
| |
| osutils.SafeUnlink(self.lock_file) |
| |
| def _HelperInsideProcess(self, blocking, shared, locktype=locking.LOCKF, |
| blocking_timeout=None): |
| """Helper method that runs a basic test with/without blocking.""" |
| try: |
| lock = locking.FileLock( |
| self.lock_file, blocking=blocking, locktype=locktype, |
| blocking_timeout=blocking_timeout) |
| with lock.lock(shared): |
| pass |
| sys.exit(LOCK_ACQUIRED) |
| except locking.LockNotAcquiredError: |
| sys.exit(LOCK_NOT_ACQUIRED) |
| |
| def _HelperStartProcess(self, blocking=False, shared=False, |
| locktype=locking.LOCKF, |
| blocking_timeout=None): |
| """Create a process and invoke _HelperInsideProcess in it.""" |
| p = multiprocessing.Process(target=self._HelperInsideProcess, |
| args=(blocking, shared, locktype, |
| blocking_timeout)) |
| p.start() |
| |
| # It's highly probably that p will have tried to grab the lock before the |
| # timer expired, but not certain. |
| time.sleep(0.1) |
| |
| return p |
| |
| def _HelperWithProcess(self, expected, blocking=False, shared=False, |
| locktype=locking.LOCKF): |
| """Create a process and invoke _HelperInsideProcess in it.""" |
| p = multiprocessing.Process(target=self._HelperInsideProcess, |
| args=(blocking, shared, locktype)) |
| p.start() |
| p.join() |
| self.assertEqual(p.exitcode, expected) |
| |
| def testSingleLock(self): |
| """Just test getting releasing a lock with options.""" |
| arg_list = [ |
| [True, False], # blocking |
| [True, False], # shared |
| [locking.FLOCK, locking.LOCKF], # locking mechanism |
| ] |
| for args in itertools.product(*arg_list): |
| self._HelperSingleLockTest(*args) |
| |
| def testDoubleLockWithFlock(self): |
| """Tests that double locks do block with flock.""" |
| lock1 = locking.FileLock( |
| self.lock_file, blocking=False, locktype=locking.FLOCK) |
| lock2 = locking.FileLock( |
| self.lock_file, blocking=False, locktype=locking.FLOCK) |
| |
| with lock1.write_lock(): |
| self.assertTrue(lock1.IsLocked()) |
| self.assertFalse(lock2.IsLocked()) |
| |
| self.assertRaises(locking.LockNotAcquiredError, lock2.write_lock) |
| self.assertTrue(lock1.IsLocked()) |
| self.assertFalse(lock2.IsLocked()) |
| |
| self.assertFalse(lock1.IsLocked()) |
| self.assertFalse(lock2.IsLocked()) |
| |
| lock2.unlock() |
| self.assertFalse(lock1.IsLocked()) |
| self.assertFalse(lock2.IsLocked()) |
| |
| def testDoubleLockTimeoutWithFlock(self): |
| """Tests that double locking the same lock times out.""" |
| lock1 = locking.FileLock( |
| self.lock_file, blocking=True, locktype=locking.FLOCK, |
| blocking_timeout=1, verbose=True) |
| lock2 = locking.FileLock( |
| self.lock_file, blocking=True, locktype=locking.FLOCK, |
| blocking_timeout=1, verbose=True) |
| |
| # Verify that if we grab the first lock, trying to use lock2 |
| # (on the same file) will throw an exception. |
| with lock1.write_lock(): |
| self.assertTrue(lock1.IsLocked()) |
| self.assertFalse(lock2.IsLocked()) |
| |
| self.assertRaises(timeout_util.TimeoutError, lock2.write_lock) |
| self.assertTrue(lock1.IsLocked()) |
| |
| def testDoubleLockWithLockf(self): |
| """Tests that double locks don't block with lockf.""" |
| lock1 = locking.FileLock( |
| self.lock_file, blocking=False, locktype=locking.LOCKF) |
| lock2 = locking.FileLock( |
| self.lock_file, blocking=False, locktype=locking.LOCKF) |
| with lock1.write_lock(): |
| self.assertTrue(lock1.IsLocked()) |
| self.assertFalse(lock2.IsLocked()) |
| |
| # With lockf, we can lock the same file twice in the same process. |
| with lock2.write_lock(): |
| self.assertTrue(lock1.IsLocked()) |
| self.assertTrue(lock2.IsLocked()) |
| |
| self.assertFalse(lock1.IsLocked()) |
| self.assertFalse(lock2.IsLocked()) |
| |
| def testContextMgr(self): |
| """Make sure we behave properly with 'with'.""" |
| # Create an instance, and use it in a with. |
| prelock = locking.FileLock(self.lock_file) |
| self._HelperWithProcess(expected=LOCK_ACQUIRED) |
| |
| with prelock.write_lock() as lock: |
| # Assert the instance didn't change. |
| self.assertIs(prelock, lock) |
| self._HelperWithProcess(expected=LOCK_NOT_ACQUIRED) |
| |
| self._HelperWithProcess(expected=LOCK_ACQUIRED) |
| |
| # Construct the instance in the with expression. |
| with locking.FileLock(self.lock_file).write_lock() as lock: |
| self.assertIsInstance(lock, locking.FileLock) |
| self._HelperWithProcess(expected=LOCK_NOT_ACQUIRED) |
| |
| self._HelperWithProcess(expected=LOCK_ACQUIRED) |
| |
| def testAcquireBeforeWith(self): |
| """Sometimes you want to grab a lock and then return it into 'with'.""" |
| lock = locking.FileLock(self.lock_file, blocking=False) |
| |
| lock.write_lock() |
| self._HelperWithProcess(expected=LOCK_NOT_ACQUIRED) |
| |
| with lock: |
| self._HelperWithProcess(expected=LOCK_NOT_ACQUIRED) |
| |
| self._HelperWithProcess(expected=LOCK_ACQUIRED) |
| |
| def testSingleProcessLock(self): |
| """Test grabbing the same lock in processes with no conflicts.""" |
| arg_list = [ |
| [LOCK_ACQUIRED], |
| [True, False], # blocking |
| [True, False], # shared |
| [locking.FLOCK, locking.LOCKF], # locking mechanism |
| ] |
| for args in itertools.product(*arg_list): |
| self._HelperWithProcess(*args) |
| |
| def testNonBlockingConflicts(self): |
| """Test that we get a lock conflict for non-blocking locks.""" |
| with locking.FileLock(self.lock_file).write_lock(): |
| self._HelperWithProcess(expected=LOCK_NOT_ACQUIRED) |
| |
| self._HelperWithProcess(expected=LOCK_NOT_ACQUIRED, shared=True) |
| |
| # Can grab it after it's released. |
| self._HelperWithProcess(expected=LOCK_ACQUIRED) |
| |
| def testSharedLocks(self): |
| """Test lock conflict for blocking locks.""" |
| # Intial lock is NOT shared. |
| with locking.FileLock(self.lock_file).write_lock(): |
| self._HelperWithProcess(expected=LOCK_NOT_ACQUIRED, shared=True) |
| |
| # Intial lock IS shared. |
| with locking.FileLock(self.lock_file).read_lock(): |
| self._HelperWithProcess(expected=LOCK_ACQUIRED, shared=True) |
| self._HelperWithProcess(expected=LOCK_NOT_ACQUIRED, |
| shared=False) |
| |
| def testBlockingConflicts(self): |
| """Test lock conflict for blocking locks.""" |
| # Intial lock is blocking, exclusive. |
| with locking.FileLock(self.lock_file, blocking=True).write_lock(): |
| self._HelperWithProcess(expected=LOCK_NOT_ACQUIRED, blocking=False) |
| |
| p = self._HelperStartProcess(blocking=True, shared=False) |
| |
| # when the with clause exits, p should unblock and get the lock, setting |
| # its exit code to sucess now. |
| p.join() |
| self.assertEqual(p.exitcode, LOCK_ACQUIRED) |
| |
| # Intial lock is NON blocking. |
| with locking.FileLock(self.lock_file, blocking=False).write_lock(): |
| self._HelperWithProcess(expected=LOCK_NOT_ACQUIRED) |
| |
| p = self._HelperStartProcess(blocking=True, shared=False) |
| |
| # when the with clause exits, p should unblock and get the lock, setting |
| # it's exit code to sucess now. |
| p.join() |
| self.assertEqual(p.exitcode, LOCK_ACQUIRED) |
| |
| # Intial lock is shared, blocking lock is exclusive. |
| with locking.FileLock(self.lock_file, blocking=False).read_lock(): |
| self._HelperWithProcess(expected=LOCK_NOT_ACQUIRED) |
| self._HelperWithProcess(expected=LOCK_ACQUIRED, shared=True) |
| |
| p = self._HelperStartProcess(blocking=True, shared=False) |
| q = self._HelperStartProcess(blocking=True, shared=False) |
| |
| # when the with clause exits, p should unblock and get the lock, setting |
| # it's exit code to sucess now. |
| p.join() |
| self.assertEqual(p.exitcode, LOCK_ACQUIRED) |
| q.join() |
| self.assertEqual(p.exitcode, LOCK_ACQUIRED) |
| |
| |
| class PortableLinkLockTest(cros_test_lib.TempDirTestCase): |
| """Test locking.PortableLinkLock class.""" |
| |
| def tearDown(self): |
| """Looks for leaked files from the locking process.""" |
| leaked_files = os.listdir(self.tempdir) |
| self.assertFalse(leaked_files, |
| 'Found unexpected leaked files from locking: %r' % |
| leaked_files) |
| |
| def testLockExclusivity(self): |
| """Test that when we have a lock, someone else can't grab it.""" |
| lock_path = os.path.join(self.tempdir, 'locked_file') |
| with locking.PortableLinkLock(lock_path, max_retry=0): |
| with self.assertRaises(locking.LockNotAcquiredError): |
| with locking.PortableLinkLock(lock_path, max_retry=5, sleep=0.1): |
| self.fail('We acquired a lock twice?') |
| |
| def testCanUnlock(self): |
| """Test that we release locks correctly.""" |
| lock_path = os.path.join(self.tempdir, 'locked_file') |
| with locking.PortableLinkLock(lock_path, max_retry=0): |
| pass |
| with locking.PortableLinkLock(lock_path, max_retry=0): |
| pass |
| |
| |
| class PipeLockTest(cros_test_lib.TestCase): |
| """Test locking.PipeLock class.""" |
| |
| def testFdLeakage(self): |
| """Make sure we don't leak any fds.""" |
| fds_before = os.listdir('/proc/self/fd/') |
| lock = locking.PipeLock() |
| fds_after = os.listdir('/proc/self/fd/') |
| self.assertEqual(len(fds_before), len(fds_after) - 2) |
| del lock |
| fds_finished = os.listdir('/proc/self/fd/') |
| self.assertEqual(fds_before, fds_finished) |
| |
| def testSimple(self): |
| """Test we can Wait/Post.""" |
| # If this fails, we'd just hang :). |
| with timeout_util.Timeout(30): |
| lock = locking.PipeLock() |
| lock.Post() |
| lock.Post() |
| lock.Wait() |
| lock.Wait() |
| del lock |
| |
| def testParallel(self): |
| """Test interprocesses actually sync.""" |
| write_lock = locking.PipeLock() |
| read_lock = locking.PipeLock() |
| |
| with osutils.TempDir() as tempdir: |
| # Let the child create a file, but make sure the parent holds us off. |
| # Then make the parent wait for the child to tell us it's done. |
| flag_file = os.path.join(tempdir, 'foo') |
| |
| pid = os.fork() |
| if pid == 0: |
| # Child. |
| # pylint: disable=protected-access |
| try: |
| write_lock.Wait() |
| del write_lock |
| osutils.Touch(flag_file) |
| read_lock.Post() |
| del read_lock |
| except Exception: |
| os._exit(1) |
| finally: |
| # No matter what happens, we must exit w/out running handlers. |
| os._exit(0) |
| else: |
| # Parent. |
| time.sleep(0.5) |
| self.assertNotExists(flag_file) |
| write_lock.Post() |
| del write_lock |
| read_lock.Wait() |
| del read_lock |
| self.assertExists(flag_file) |
| |
| status = os.waitpid(pid, 0)[1] |
| self.assertEqual(process_util.GetExitStatus(status), 0) |
| |
| def testParallelMany(self): |
| """Same as testParallel, but with many more processes for stressing.""" |
| parallel.RunParallelSteps([self.testParallel] * 40) |
