From 74998795d1bde2b62ca5362f71d212285eb1be4d Mon Sep 17 00:00:00 2001 From: Michael Still Date: Tue, 23 Oct 2012 14:25:25 -0700 Subject: [PATCH] Migrate to fileutils and lockutils. Migrate nova to using openstack-common's file and lock utilities. Resolves bug 1063230. Change-Id: I1a4c87856bc08cd33b61d7098ed856baa4583654 --- doc/source/devref/threading.rst | 2 +- nova/flags.py | 3 - nova/openstack/common/fileutils.py | 35 +++++ nova/openstack/common/lockutils.py | 233 +++++++++++++++++++++++++++++ nova/tests/test_imagebackend.py | 13 +- nova/tests/test_libvirt.py | 3 +- nova/tests/test_misc.py | 94 ------------ nova/utils.py | 192 ------------------------ openstack-common.conf | 2 +- 9 files changed, 280 insertions(+), 297 deletions(-) create mode 100644 nova/openstack/common/fileutils.py create mode 100644 nova/openstack/common/lockutils.py diff --git a/doc/source/devref/threading.rst b/doc/source/devref/threading.rst index 356ca2f6f..f1ab0cc06 100644 --- a/doc/source/devref/threading.rst +++ b/doc/source/devref/threading.rst @@ -12,7 +12,7 @@ view, each OpenStack service runs in a single thread. The use of green threads reduces the likelihood of race conditions, but does not completely eliminate them. In some cases, you may need to use the -``@utils.synchronized(...)`` decorator to avoid races. +``@lockutils.synchronized(...)`` decorator to avoid races. In addition, since there is only one operating system thread, a call that blocks that main thread will block the entire process. diff --git a/nova/flags.py b/nova/flags.py index e9057d8db..639738fe7 100644 --- a/nova/flags.py +++ b/nova/flags.py @@ -91,9 +91,6 @@ core_opts = [ cfg.StrOpt('state_path', default='$pybasedir', help="Top-level directory for maintaining nova's state"), - cfg.StrOpt('lock_path', - default='$pybasedir', - help='Directory to use for lock files'), ] debug_opts = [ diff --git a/nova/openstack/common/fileutils.py b/nova/openstack/common/fileutils.py new file mode 100644 index 000000000..4746ad498 --- /dev/null +++ b/nova/openstack/common/fileutils.py @@ -0,0 +1,35 @@ +# vim: tabstop=4 shiftwidth=4 softtabstop=4 + +# Copyright 2011 OpenStack LLC. +# All Rights Reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"); you may +# not use this file except in compliance with the License. You may obtain +# a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +# License for the specific language governing permissions and limitations +# under the License. + + +import errno +import os + + +def ensure_tree(path): + """Create a directory (and any ancestor directories required) + + :param path: Directory to create + """ + try: + os.makedirs(path) + except OSError as exc: + if exc.errno == errno.EEXIST: + if not os.path.isdir(path): + raise + else: + raise diff --git a/nova/openstack/common/lockutils.py b/nova/openstack/common/lockutils.py new file mode 100644 index 000000000..2840ce6f7 --- /dev/null +++ b/nova/openstack/common/lockutils.py @@ -0,0 +1,233 @@ +# vim: tabstop=4 shiftwidth=4 softtabstop=4 + +# Copyright 2011 OpenStack LLC. +# All Rights Reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"); you may +# not use this file except in compliance with the License. You may obtain +# a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +# License for the specific language governing permissions and limitations +# under the License. + + +import errno +import functools +import os +import shutil +import tempfile +import time +import weakref + +from eventlet import greenthread +from eventlet import semaphore + +from nova.openstack.common import cfg +from nova.openstack.common import fileutils +from nova.openstack.common import log as logging + + +LOG = logging.getLogger(__name__) + + +util_opts = [ + cfg.BoolOpt('disable_process_locking', default=False, + help='Whether to disable inter-process locks'), + cfg.StrOpt('lock_path', + default=os.path.abspath(os.path.join(os.path.dirname(__file__), + '../')), + help='Directory to use for lock files') +] + + +CONF = cfg.CONF +CONF.register_opts(util_opts) + + +class _InterProcessLock(object): + """Lock implementation which allows multiple locks, working around + issues like bugs.debian.org/cgi-bin/bugreport.cgi?bug=632857 and does + not require any cleanup. Since the lock is always held on a file + descriptor rather than outside of the process, the lock gets dropped + automatically if the process crashes, even if __exit__ is not executed. + + There are no guarantees regarding usage by multiple green threads in a + single process here. This lock works only between processes. Exclusive + access between local threads should be achieved using the semaphores + in the @synchronized decorator. + + Note these locks are released when the descriptor is closed, so it's not + safe to close the file descriptor while another green thread holds the + lock. Just opening and closing the lock file can break synchronisation, + so lock files must be accessed only using this abstraction. + """ + + def __init__(self, name): + self.lockfile = None + self.fname = name + + def __enter__(self): + self.lockfile = open(self.fname, 'w') + + while True: + try: + # Using non-blocking locks since green threads are not + # patched to deal with blocking locking calls. + # Also upon reading the MSDN docs for locking(), it seems + # to have a laughable 10 attempts "blocking" mechanism. + self.trylock() + return self + except IOError, e: + if e.errno in (errno.EACCES, errno.EAGAIN): + # external locks synchronise things like iptables + # updates - give it some time to prevent busy spinning + time.sleep(0.01) + else: + raise + + def __exit__(self, exc_type, exc_val, exc_tb): + try: + self.unlock() + self.lockfile.close() + except IOError: + LOG.exception(_("Could not release the acquired lock `%s`"), + self.fname) + + def trylock(self): + raise NotImplementedError() + + def unlock(self): + raise NotImplementedError() + + +class _WindowsLock(_InterProcessLock): + def trylock(self): + msvcrt.locking(self.lockfile, msvcrt.LK_NBLCK, 1) + + def unlock(self): + msvcrt.locking(self.lockfile, msvcrt.LK_UNLCK, 1) + + +class _PosixLock(_InterProcessLock): + def trylock(self): + fcntl.lockf(self.lockfile, fcntl.LOCK_EX | fcntl.LOCK_NB) + + def unlock(self): + fcntl.lockf(self.lockfile, fcntl.LOCK_UN) + + +if os.name == 'nt': + import msvcrt + InterProcessLock = _WindowsLock +else: + import fcntl + InterProcessLock = _PosixLock + +_semaphores = weakref.WeakValueDictionary() + + +def synchronized(name, lock_file_prefix, external=False, lock_path=None): + """Synchronization decorator. + + Decorating a method like so:: + + @synchronized('mylock') + def foo(self, *args): + ... + + ensures that only one thread will execute the bar method at a time. + + Different methods can share the same lock:: + + @synchronized('mylock') + def foo(self, *args): + ... + + @synchronized('mylock') + def bar(self, *args): + ... + + This way only one of either foo or bar can be executing at a time. + + The lock_file_prefix argument is used to provide lock files on disk with a + meaningful prefix. The prefix should end with a hyphen ('-') if specified. + + The external keyword argument denotes whether this lock should work across + multiple processes. This means that if two different workers both run a + a method decorated with @synchronized('mylock', external=True), only one + of them will execute at a time. + + The lock_path keyword argument is used to specify a special location for + external lock files to live. If nothing is set, then CONF.lock_path is + used as a default. + """ + + def wrap(f): + @functools.wraps(f) + def inner(*args, **kwargs): + # NOTE(soren): If we ever go natively threaded, this will be racy. + # See http://stackoverflow.com/questions/5390569/dyn + # amically-allocating-and-destroying-mutexes + sem = _semaphores.get(name, semaphore.Semaphore()) + if name not in _semaphores: + # this check is not racy - we're already holding ref locally + # so GC won't remove the item and there was no IO switch + # (only valid in greenthreads) + _semaphores[name] = sem + + with sem: + LOG.debug(_('Got semaphore "%(lock)s" for method ' + '"%(method)s"...'), {'lock': name, + 'method': f.__name__}) + if external and not CONF.disable_process_locking: + LOG.debug(_('Attempting to grab file lock "%(lock)s" for ' + 'method "%(method)s"...'), + {'lock': name, 'method': f.__name__}) + cleanup_dir = False + + # We need a copy of lock_path because it is non-local + local_lock_path = lock_path + if not local_lock_path: + local_lock_path = CONF.lock_path + + if not local_lock_path: + cleanup_dir = True + local_lock_path = tempfile.mkdtemp() + + if not os.path.exists(local_lock_path): + cleanup_dir = True + fileutils.ensure_tree(local_lock_path) + + # NOTE(mikal): the lock name cannot contain directory + # separators + safe_name = name.replace(os.sep, '_') + lock_file_name = '%s%s' % (lock_file_prefix, safe_name) + lock_file_path = os.path.join(local_lock_path, + lock_file_name) + + try: + lock = InterProcessLock(lock_file_path) + with lock: + LOG.debug(_('Got file lock "%(lock)s" at %(path)s ' + 'for method "%(method)s"...'), + {'lock': name, + 'path': lock_file_path, + 'method': f.__name__}) + retval = f(*args, **kwargs) + finally: + # NOTE(vish): This removes the tempdir if we needed + # to create one. This is used to cleanup + # the locks left behind by unit tests. + if cleanup_dir: + shutil.rmtree(local_lock_path) + else: + retval = f(*args, **kwargs) + + return retval + return inner + return wrap diff --git a/nova/tests/test_imagebackend.py b/nova/tests/test_imagebackend.py index f0bb71829..088cb0a82 100644 --- a/nova/tests/test_imagebackend.py +++ b/nova/tests/test_imagebackend.py @@ -18,6 +18,7 @@ import os from nova import flags +from nova.openstack.common import fileutils from nova import test from nova.tests import fake_libvirt_utils from nova.virt.libvirt import imagebackend @@ -56,8 +57,8 @@ class _ImageTestCase(test.TestCase): os.path.exists(self.TEMPLATE_PATH).AndReturn(False) fn = self.mox.CreateMockAnything() fn(target=self.TEMPLATE_PATH) - self.mox.StubOutWithMock(imagebackend.utils, 'ensure_tree') - imagebackend.utils.ensure_tree(self.TEMPLATE_DIR) + self.mox.StubOutWithMock(imagebackend.fileutils, 'ensure_tree') + imagebackend.fileutils.ensure_tree(self.TEMPLATE_DIR) self.mox.ReplayAll() image = self.image_class(self.INSTANCE, self.NAME) @@ -83,7 +84,7 @@ class _ImageTestCase(test.TestCase): os.path.exists(self.TEMPLATE_PATH).AndReturn(False) fn = self.mox.CreateMockAnything() fn(target=self.TEMPLATE_PATH) - self.mox.StubOutWithMock(imagebackend.utils, 'ensure_tree') + self.mox.StubOutWithMock(imagebackend.fileutils, 'ensure_tree') self.mox.ReplayAll() image = self.image_class(self.INSTANCE, self.NAME) @@ -117,7 +118,8 @@ class RawTestCase(_ImageTestCase): def prepare_mocks(self): fn = self.mox.CreateMockAnything() - self.mox.StubOutWithMock(imagebackend.utils.synchronized, '__call__') + self.mox.StubOutWithMock(imagebackend.lockutils.synchronized, + '__call__') self.mox.StubOutWithMock(imagebackend.libvirt_utils, 'copy_image') self.mox.StubOutWithMock(imagebackend.disk, 'extend') return fn @@ -167,7 +169,8 @@ class Qcow2TestCase(_ImageTestCase): def prepare_mocks(self): fn = self.mox.CreateMockAnything() - self.mox.StubOutWithMock(imagebackend.utils.synchronized, '__call__') + self.mox.StubOutWithMock(imagebackend.lockutils.synchronized, + '__call__') self.mox.StubOutWithMock(imagebackend.libvirt_utils, 'create_cow_image') self.mox.StubOutWithMock(imagebackend.libvirt_utils, 'copy_image') diff --git a/nova/tests/test_libvirt.py b/nova/tests/test_libvirt.py index 7e73516b0..c84440e58 100644 --- a/nova/tests/test_libvirt.py +++ b/nova/tests/test_libvirt.py @@ -37,6 +37,7 @@ from nova import context from nova import db from nova import exception from nova import flags +from nova.openstack.common import fileutils from nova.openstack.common import importutils from nova.openstack.common import jsonutils from nova.openstack.common import log as logging @@ -448,7 +449,7 @@ class CacheConcurrencyTestCase(test.TestCase): # use for tests. So, create the path here so utils.synchronized() # won't delete it out from under one of the threads. self.lock_path = os.path.join(FLAGS.instances_path, 'locks') - utils.ensure_tree(self.lock_path) + fileutils.ensure_tree(self.lock_path) def fake_exists(fname): basedir = os.path.join(FLAGS.instances_path, FLAGS.base_dir_name) diff --git a/nova/tests/test_misc.py b/nova/tests/test_misc.py index 10443ecde..4e989f315 100644 --- a/nova/tests/test_misc.py +++ b/nova/tests/test_misc.py @@ -24,7 +24,6 @@ from eventlet import greenthread from nova import exception from nova import test -from nova import utils class ExceptionTestCase(test.TestCase): @@ -63,96 +62,3 @@ class ProjectTestCase(test.TestCase): helpful_msg = (_("The following migrations are missing a downgrade:" "\n\t%s") % '\n\t'.join(sorted(missing_downgrade))) self.assert_(not missing_downgrade, helpful_msg) - - -class LockTestCase(test.TestCase): - def test_synchronized_wrapped_function_metadata(self): - @utils.synchronized('whatever') - def foo(): - """Bar""" - pass - self.assertEquals(foo.__doc__, 'Bar', "Wrapped function's docstring " - "got lost") - self.assertEquals(foo.__name__, 'foo', "Wrapped function's name " - "got mangled") - - def test_synchronized_internally(self): - """We can lock across multiple green threads""" - saved_sem_num = len(utils._semaphores) - seen_threads = list() - - @utils.synchronized('testlock2', external=False) - def f(id): - for x in range(10): - seen_threads.append(id) - greenthread.sleep(0) - - threads = [] - pool = greenpool.GreenPool(10) - for i in range(10): - threads.append(pool.spawn(f, i)) - - for thread in threads: - thread.wait() - - self.assertEquals(len(seen_threads), 100) - # Looking at the seen threads, split it into chunks of 10, and verify - # that the last 9 match the first in each chunk. - for i in range(10): - for j in range(9): - self.assertEquals(seen_threads[i * 10], - seen_threads[i * 10 + 1 + j]) - - self.assertEqual(saved_sem_num, len(utils._semaphores), - "Semaphore leak detected") - - def test_nested_external_works(self): - """We can nest external syncs""" - with utils.tempdir() as tempdir: - self.flags(lock_path=tempdir) - sentinel = object() - - @utils.synchronized('testlock1', external=True) - def outer_lock(): - - @utils.synchronized('testlock2', external=True) - def inner_lock(): - return sentinel - return inner_lock() - - self.assertEqual(sentinel, outer_lock()) - - def test_synchronized_externally(self): - """We can lock across multiple processes""" - with utils.tempdir() as tempdir: - self.flags(lock_path=tempdir) - rpipe1, wpipe1 = os.pipe() - rpipe2, wpipe2 = os.pipe() - - @utils.synchronized('testlock1', external=True) - def f(rpipe, wpipe): - try: - os.write(wpipe, "foo") - except OSError, e: - self.assertEquals(e.errno, errno.EPIPE) - return - - rfds, _wfds, _efds = select.select([rpipe], [], [], 1) - self.assertEquals(len(rfds), 0, "The other process, which was" - " supposed to be locked, " - "wrote on its end of the " - "pipe") - os.close(rpipe) - - pid = os.fork() - if pid > 0: - os.close(wpipe1) - os.close(rpipe2) - - f(rpipe1, wpipe2) - else: - os.close(rpipe1) - os.close(wpipe2) - - f(rpipe2, wpipe1) - os._exit(0) diff --git a/nova/utils.py b/nova/utils.py index c69b97ed8..3cca396c0 100644 --- a/nova/utils.py +++ b/nova/utils.py @@ -586,183 +586,6 @@ def utf8(value): return value -class _InterProcessLock(object): - """Lock implementation which allows multiple locks, working around - issues like bugs.debian.org/cgi-bin/bugreport.cgi?bug=632857 and does - not require any cleanup. Since the lock is always held on a file - descriptor rather than outside of the process, the lock gets dropped - automatically if the process crashes, even if __exit__ is not executed. - - There are no guarantees regarding usage by multiple green threads in a - single process here. This lock works only between processes. Exclusive - access between local threads should be achieved using the semaphores - in the @synchronized decorator. - - Note these locks are released when the descriptor is closed, so it's not - safe to close the file descriptor while another green thread holds the - lock. Just opening and closing the lock file can break synchronisation, - so lock files must be accessed only using this abstraction. - """ - - def __init__(self, name): - self.lockfile = None - self.fname = name - - def __enter__(self): - self.lockfile = open(self.fname, 'w') - - while True: - try: - # Using non-blocking locks since green threads are not - # patched to deal with blocking locking calls. - # Also upon reading the MSDN docs for locking(), it seems - # to have a laughable 10 attempts "blocking" mechanism. - self.trylock() - return self - except IOError, e: - if e.errno in (errno.EACCES, errno.EAGAIN): - # external locks synchronise things like iptables - # updates - give it some time to prevent busy spinning - time.sleep(0.01) - else: - raise - - def __exit__(self, exc_type, exc_val, exc_tb): - try: - self.unlock() - self.lockfile.close() - except IOError: - LOG.exception(_("Could not release the acquired lock `%s`") - % self.fname) - - def trylock(self): - raise NotImplementedError() - - def unlock(self): - raise NotImplementedError() - - -class _WindowsLock(_InterProcessLock): - def trylock(self): - msvcrt.locking(self.lockfile, msvcrt.LK_NBLCK, 1) - - def unlock(self): - msvcrt.locking(self.lockfile, msvcrt.LK_UNLCK, 1) - - -class _PosixLock(_InterProcessLock): - def trylock(self): - fcntl.lockf(self.lockfile, fcntl.LOCK_EX | fcntl.LOCK_NB) - - def unlock(self): - fcntl.lockf(self.lockfile, fcntl.LOCK_UN) - - -if os.name == 'nt': - import msvcrt - InterProcessLock = _WindowsLock -else: - import fcntl - InterProcessLock = _PosixLock - -_semaphores = weakref.WeakValueDictionary() - - -def synchronized(name, external=False, lock_path=None): - """Synchronization decorator. - - Decorating a method like so:: - - @synchronized('mylock') - def foo(self, *args): - ... - - ensures that only one thread will execute the bar method at a time. - - Different methods can share the same lock:: - - @synchronized('mylock') - def foo(self, *args): - ... - - @synchronized('mylock') - def bar(self, *args): - ... - - This way only one of either foo or bar can be executing at a time. - - The external keyword argument denotes whether this lock should work across - multiple processes. This means that if two different workers both run a - a method decorated with @synchronized('mylock', external=True), only one - of them will execute at a time. - - The lock_path keyword argument is used to specify a special location for - external lock files to live. If nothing is set, then FLAGS.lock_path is - used as a default. - """ - - def wrap(f): - @functools.wraps(f) - def inner(*args, **kwargs): - # NOTE(soren): If we ever go natively threaded, this will be racy. - # See http://stackoverflow.com/questions/5390569/dyn - # amically-allocating-and-destroying-mutexes - sem = _semaphores.get(name, semaphore.Semaphore()) - if name not in _semaphores: - # this check is not racy - we're already holding ref locally - # so GC won't remove the item and there was no IO switch - # (only valid in greenthreads) - _semaphores[name] = sem - - with sem: - LOG.debug(_('Got semaphore "%(lock)s" for method ' - '"%(method)s"...'), {'lock': name, - 'method': f.__name__}) - if external and not FLAGS.disable_process_locking: - LOG.debug(_('Attempting to grab file lock "%(lock)s" for ' - 'method "%(method)s"...'), - {'lock': name, 'method': f.__name__}) - cleanup_dir = False - - # We need a copy of lock_path because it is non-local - local_lock_path = lock_path - if not local_lock_path: - local_lock_path = FLAGS.lock_path - - if not local_lock_path: - cleanup_dir = True - local_lock_path = tempfile.mkdtemp() - - if not os.path.exists(local_lock_path): - cleanup_dir = True - ensure_tree(local_lock_path) - - # NOTE(mikal): the lock name cannot contain directory - # separators - safe_name = name.replace(os.sep, '_') - lock_file_path = os.path.join(local_lock_path, - 'nova-%s' % safe_name) - try: - lock = InterProcessLock(lock_file_path) - with lock: - LOG.debug(_('Got file lock "%(lock)s" for ' - 'method "%(method)s"...'), - {'lock': name, 'method': f.__name__}) - retval = f(*args, **kwargs) - finally: - # NOTE(vish): This removes the tempdir if we needed - # to create one. This is used to cleanup - # the locks left behind by unit tests. - if cleanup_dir: - shutil.rmtree(local_lock_path) - else: - retval = f(*args, **kwargs) - - return retval - return inner - return wrap - - def delete_if_exists(pathname): """delete a file, but ignore file not found error""" @@ -1314,21 +1137,6 @@ class UndoManager(object): self._rollback() -def ensure_tree(path): - """Create a directory (and any ancestor directories required) - - :param path: Directory to create - """ - try: - os.makedirs(path) - except OSError as exc: - if exc.errno == errno.EEXIST: - if not os.path.isdir(path): - raise - else: - raise - - def mkfs(fs, path, label=None): """Format a file or block device diff --git a/openstack-common.conf b/openstack-common.conf index 398bfb4ae..9a09fd5a8 100644 --- a/openstack-common.conf +++ b/openstack-common.conf @@ -1,7 +1,7 @@ [DEFAULT] # The list of modules to copy from openstack-common -modules=cfg,context,excutils,gettextutils,importutils,iniparser,jsonutils,local,log,network_utils,notifier,plugin,policy,setup,timeutils,rpc +modules=cfg,context,excutils,fileutils,gettextutils,importutils,iniparser,jsonutils,local,lockutils,log,network_utils,notifier,plugin,policy,setup,timeutils,rpc # The base module to hold the copy of openstack.common base=nova