This is only used on the compute and network side in Nova.
Change-Id: I6211c32f585b1b5ff66d1a7a5f65dd25857d76f0
Reviewed-on: https://review.openstack.org/9888
Reviewed-by: Vish Ishaya <vishvananda@gmail.com>
Approved: John Griffith <john.griffith@solidfire.com>
Tested-by: Jenkins
cfg.StrOpt('state_path',
default='$pybasedir',
help="Top-level directory for maintaining cinder's state"),
- cfg.StrOpt('lock_path',
- default='$pybasedir',
- help='Directory to use for lock files'),
]
debug_opts = [
vcs_string = version.version_string_with_vcs()
LOG.audit(_('Starting %(topic)s node (version %(vcs_string)s)'),
{'topic': self.topic, 'vcs_string': vcs_string})
- utils.cleanup_file_locks()
rpc.register_opts(FLAGS)
self.manager.init_host()
self.model_disconnected = False
:returns: None
"""
- utils.cleanup_file_locks()
rpc.register_opts(FLAGS)
if self.manager:
self.manager.init_host()
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_fails(self):
- """We can not nest external syncs"""
-
- @utils.synchronized('testlock1', external=True)
- def outer_lock():
-
- @utils.synchronized('testlock2', external=True)
- def inner_lock():
- pass
- inner_lock()
- try:
- self.assertRaises(lockfile.NotMyLock, outer_lock)
- finally:
- utils.cleanup_file_locks()
-
- def test_synchronized_externally(self):
- """We can lock across multiple processes"""
- 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)
self.assertEquals(h1, h2)
-class TestGreenLocks(test.TestCase):
- def test_concurrent_green_lock_succeeds(self):
- """Verify spawn_n greenthreads with two locks run concurrently.
-
- This succeeds with spawn but fails with spawn_n because lockfile
- gets the same thread id for both spawn_n threads. Our workaround
- of using the GreenLockFile will work even if the issue is fixed.
- """
- self.completed = False
- with utils.tempdir() as tmpdir:
-
- def locka(wait):
- a = utils.GreenLockFile(os.path.join(tmpdir, 'a'))
- a.acquire()
- wait.wait()
- a.release()
- self.completed = True
-
- def lockb(wait):
- b = utils.GreenLockFile(os.path.join(tmpdir, 'b'))
- b.acquire()
- wait.wait()
- b.release()
-
- wait1 = eventlet.event.Event()
- wait2 = eventlet.event.Event()
- pool = greenpool.GreenPool()
- pool.spawn_n(locka, wait1)
- pool.spawn_n(lockb, wait2)
- wait2.send()
- eventlet.sleep(0)
- wait1.send()
- pool.waitall()
- self.assertTrue(self.completed)
-
-
-class TestLockCleanup(test.TestCase):
- """unit tests for utils.cleanup_file_locks()"""
-
- def setUp(self):
- super(TestLockCleanup, self).setUp()
-
- self.pid = os.getpid()
- self.dead_pid = self._get_dead_pid()
- self.tempdir = tempfile.mkdtemp()
- self.flags(lock_path=self.tempdir)
- self.lock_name = 'cinder-testlock'
- self.lock_file = os.path.join(FLAGS.lock_path,
- self.lock_name + '.lock')
- self.hostname = socket.gethostname()
- print self.pid, self.dead_pid
- try:
- os.unlink(self.lock_file)
- except OSError as (errno, strerror):
- if errno == 2:
- pass
-
- def tearDown(self):
- shutil.rmtree(self.tempdir)
- super(TestLockCleanup, self).tearDown()
-
- def _get_dead_pid(self):
- """get a pid for a process that does not exist"""
-
- candidate_pid = self.pid - 1
- while os.path.exists(os.path.join('/proc', str(candidate_pid))):
- candidate_pid -= 1
- if candidate_pid == 1:
- return 0
- return candidate_pid
-
- def _get_sentinel_name(self, hostname, pid, thread='MainThread'):
- return os.path.join(FLAGS.lock_path,
- '%s.%s-%d' % (hostname, thread, pid))
-
- def _create_sentinel(self, hostname, pid, thread='MainThread'):
- name = self._get_sentinel_name(hostname, pid, thread)
- open(name, 'wb').close()
- return name
-
- def test_clean_stale_locks(self):
- """verify locks for dead processes are cleaned up"""
-
- # create sentinels for two processes, us and a 'dead' one
- # no active lock
- sentinel1 = self._create_sentinel(self.hostname, self.pid)
- sentinel2 = self._create_sentinel(self.hostname, self.dead_pid)
-
- utils.cleanup_file_locks()
-
- self.assertTrue(os.path.exists(sentinel1))
- self.assertFalse(os.path.exists(self.lock_file))
- self.assertFalse(os.path.exists(sentinel2))
-
- os.unlink(sentinel1)
-
- def test_clean_stale_locks_active(self):
- """verify locks for dead processes are cleaned with an active lock """
-
- # create sentinels for two processes, us and a 'dead' one
- # create an active lock for us
- sentinel1 = self._create_sentinel(self.hostname, self.pid)
- sentinel2 = self._create_sentinel(self.hostname, self.dead_pid)
- os.link(sentinel1, self.lock_file)
-
- utils.cleanup_file_locks()
-
- self.assertTrue(os.path.exists(sentinel1))
- self.assertTrue(os.path.exists(self.lock_file))
- self.assertFalse(os.path.exists(sentinel2))
-
- os.unlink(sentinel1)
- os.unlink(self.lock_file)
-
- def test_clean_stale_with_threads(self):
- """verify locks for multiple threads are cleaned up """
-
- # create sentinels for four threads in our process, and a 'dead'
- # process. no lock.
- sentinel1 = self._create_sentinel(self.hostname, self.pid, 'Default-1')
- sentinel2 = self._create_sentinel(self.hostname, self.pid, 'Default-2')
- sentinel3 = self._create_sentinel(self.hostname, self.pid, 'Default-3')
- sentinel4 = self._create_sentinel(self.hostname, self.pid, 'Default-4')
- sentinel5 = self._create_sentinel(self.hostname, self.dead_pid,
- 'Default-1')
-
- utils.cleanup_file_locks()
-
- self.assertTrue(os.path.exists(sentinel1))
- self.assertTrue(os.path.exists(sentinel2))
- self.assertTrue(os.path.exists(sentinel3))
- self.assertTrue(os.path.exists(sentinel4))
- self.assertFalse(os.path.exists(self.lock_file))
- self.assertFalse(os.path.exists(sentinel5))
-
- os.unlink(sentinel1)
- os.unlink(sentinel2)
- os.unlink(sentinel3)
- os.unlink(sentinel4)
-
- def test_clean_stale_with_threads_active(self):
- """verify locks for multiple threads are cleaned up """
-
- # create sentinels for four threads in our process, and a 'dead'
- # process
- sentinel1 = self._create_sentinel(self.hostname, self.pid, 'Default-1')
- sentinel2 = self._create_sentinel(self.hostname, self.pid, 'Default-2')
- sentinel3 = self._create_sentinel(self.hostname, self.pid, 'Default-3')
- sentinel4 = self._create_sentinel(self.hostname, self.pid, 'Default-4')
- sentinel5 = self._create_sentinel(self.hostname, self.dead_pid,
- 'Default-1')
-
- os.link(sentinel1, self.lock_file)
-
- utils.cleanup_file_locks()
-
- self.assertTrue(os.path.exists(sentinel1))
- self.assertTrue(os.path.exists(sentinel2))
- self.assertTrue(os.path.exists(sentinel3))
- self.assertTrue(os.path.exists(sentinel4))
- self.assertTrue(os.path.exists(self.lock_file))
- self.assertFalse(os.path.exists(sentinel5))
-
- os.unlink(sentinel1)
- os.unlink(sentinel2)
- os.unlink(sentinel3)
- os.unlink(sentinel4)
- os.unlink(self.lock_file)
-
- def test_clean_bogus_lockfiles(self):
- """verify lockfiles are cleaned """
-
- lock1 = os.path.join(FLAGS.lock_path, 'cinder-testlock1.lock')
- lock2 = os.path.join(FLAGS.lock_path, 'cinder-testlock2.lock')
- lock3 = os.path.join(FLAGS.lock_path, 'testlock3.lock')
-
- open(lock1, 'wb').close()
- open(lock2, 'wb').close()
- open(lock3, 'wb').close()
-
- utils.cleanup_file_locks()
-
- self.assertFalse(os.path.exists(lock1))
- self.assertFalse(os.path.exists(lock2))
- self.assertTrue(os.path.exists(lock3))
-
- os.unlink(lock3)
-
-
class AuditPeriodTest(test.TestCase):
def setUp(self):
import struct
import sys
import tempfile
-import threading
import time
import types
import uuid
import warnings
from xml.sax import saxutils
-from eventlet import corolocal
from eventlet import event
from eventlet import greenthread
from eventlet import semaphore
from eventlet.green import subprocess
import iso8601
-import lockfile
import netaddr
from cinder import exception
PERFECT_TIME_FORMAT = "%Y-%m-%dT%H:%M:%S.%f"
FLAGS = flags.FLAGS
-FLAGS.register_opt(
- cfg.BoolOpt('disable_process_locking', default=False,
- help='Whether to disable inter-process locks'))
-
def find_config(config_path):
"""Find a configuration file using the given hint.
anyjson.force_implementation("cinder.utils")
-class GreenLockFile(lockfile.FileLock):
- """Implementation of lockfile that allows for a lock per greenthread.
-
- Simply implements lockfile:LockBase init with an addiontall suffix
- on the unique name of the greenthread identifier
- """
- def __init__(self, path, threaded=True):
- self.path = path
- self.lock_file = os.path.abspath(path) + ".lock"
- self.hostname = socket.gethostname()
- self.pid = os.getpid()
- if threaded:
- t = threading.current_thread()
- # Thread objects in Python 2.4 and earlier do not have ident
- # attrs. Worm around that.
- ident = getattr(t, "ident", hash(t)) or hash(t)
- gident = corolocal.get_ident()
- self.tname = "-%x-%x" % (ident & 0xffffffff, gident & 0xffffffff)
- else:
- self.tname = ""
- dirname = os.path.dirname(self.lock_file)
- self.unique_name = os.path.join(dirname,
- "%s%s.%s" % (self.hostname,
- self.tname,
- self.pid))
-
-
-_semaphores = {}
-
-
-def synchronized(name, external=False):
- """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.
-
- Important limitation: you can only have one external lock running per
- thread at a time. For example the following will fail:
-
- @utils.synchronized('testlock1', external=True)
- def outer_lock():
-
- @utils.synchronized('testlock2', external=True)
- def inner_lock():
- pass
- inner_lock()
-
- outer_lock()
-
- """
-
- 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
- if name not in _semaphores:
- _semaphores[name] = semaphore.Semaphore()
- sem = _semaphores[name]
- LOG.debug(_('Attempting to grab semaphore "%(lock)s" for method '
- '"%(method)s"...'), {'lock': name,
- 'method': f.__name__})
- 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__})
- lock_file_path = os.path.join(FLAGS.lock_path,
- 'cinder-%s' % name)
- lock = GreenLockFile(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)
- else:
- retval = f(*args, **kwargs)
-
- # If no-one else is waiting for it, delete it.
- # See note about possible raciness above.
- if not sem.balance < 1:
- del _semaphores[name]
-
- return retval
- return inner
- return wrap
-
-
-def cleanup_file_locks():
- """clean up stale locks left behind by process failures
-
- The lockfile module, used by @synchronized, can leave stale lockfiles
- behind after process failure. These locks can cause process hangs
- at startup, when a process deadlocks on a lock which will never
- be unlocked.
-
- Intended to be called at service startup.
-
- """
-
- # NOTE(mikeyp) this routine incorporates some internal knowledge
- # from the lockfile module, and this logic really
- # should be part of that module.
- #
- # cleanup logic:
- # 1) look for the lockfile modules's 'sentinel' files, of the form
- # hostname.[thread-.*]-pid, extract the pid.
- # if pid doesn't match a running process, delete the file since
- # it's from a dead process.
- # 2) check for the actual lockfiles. if lockfile exists with linkcount
- # of 1, it's bogus, so delete it. A link count >= 2 indicates that
- # there are probably sentinels still linked to it from active
- # processes. This check isn't perfect, but there is no way to
- # reliably tell which sentinels refer to which lock in the
- # lockfile implementation.
-
- if FLAGS.disable_process_locking:
- return
-
- hostname = socket.gethostname()
- sentinel_re = hostname + r'\..*-(\d+$)'
- lockfile_re = r'cinder-.*\.lock'
- files = os.listdir(FLAGS.lock_path)
-
- # cleanup sentinels
- for filename in files:
- match = re.match(sentinel_re, filename)
- if match is None:
- continue
- pid = match.group(1)
- LOG.debug(_('Found sentinel %(filename)s for pid %(pid)s'),
- {'filename': filename, 'pid': pid})
- try:
- os.kill(int(pid), 0)
- except OSError, e:
- # PID wasn't found
- delete_if_exists(os.path.join(FLAGS.lock_path, filename))
- LOG.debug(_('Cleaned sentinel %(filename)s for pid %(pid)s'),
- {'filename': filename, 'pid': pid})
-
- # cleanup lock files
- for filename in files:
- match = re.match(lockfile_re, filename)
- if match is None:
- continue
- try:
- stat_info = os.stat(os.path.join(FLAGS.lock_path, filename))
- except OSError as e:
- if e.errno == errno.ENOENT:
- continue
- else:
- raise
- LOG.debug(_('Found lockfile %(file)s with link count %(count)d'),
- {'file': filename, 'count': stat_info.st_nlink})
- if stat_info.st_nlink == 1:
- delete_if_exists(os.path.join(FLAGS.lock_path, filename))
- LOG.debug(_('Cleaned lockfile %(file)s with link count %(count)d'),
- {'file': filename, 'count': stat_info.st_nlink})
-
-
def delete_if_exists(pathname):
"""delete a file, but ignore file not found error"""
# isolated_hosts=""
###### (ListOpt) Images to run on isolated host
# isolated_images=""
-###### (StrOpt) Directory to use for lock files
-# lock_path="$pybasedir"
###### (StrOpt) If this option is specified, the logging configuration file specified is used and overrides any other logging options specified. Please see the Python logging module documentation for details on logging configuration files.
# log-config=<None>
###### (StrOpt) Format string for %(asctime)s in log records. Default: %default
###### (BoolOpt) publish error events
# publish_errors=false
-######### defined in cinder.utils #########
-
-###### (BoolOpt) Whether to disable inter-process locks
-# disable_process_locking=false
-
######### defined in cinder.service #########
###### (StrOpt) The backend to use for db
+++ /dev/null
-#!/usr/bin/env python
-
-# Copyright 2012 La Honda Research Center, Inc.
-#
-# 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.
-
-"""clean_file_locks.py - Cleans stale interprocess locks
-
-This rountine can be used to find and delete stale lock files from
-cinder's interprocess synchroization. It can be used safely while services
-are running.
-
-"""
-
-import logging
-import optparse
-
-from cinder import flags
-from cinder import log
-from cinder import utils
-
-
-LOG = log.getLogger('cinder.utils')
-FLAGS = flags.FLAGS
-
-
-def parse_options():
- """process command line options."""
-
- parser = optparse.OptionParser('usage: %prog [options]')
- parser.add_option('--verbose', action='store_true',
- help='List lock files found and deleted')
-
- options, args = parser.parse_args()
-
- return options, args
-
-
-def main():
- """Main loop."""
- options, args = parse_options()
- verbose = options.verbose
-
- if verbose:
- LOG.logger.setLevel(logging.DEBUG)
- else:
- LOG.logger.setLevel(logging.INFO)
- LOG.info('Cleaning stale locks from %s' % FLAGS.lock_path)
- utils.cleanup_file_locks()
- LOG.info('Finished')
-
-if __name__ == '__main__':
- main()
Code Review / openstack-build / cinder-build.git / commitdiff