Remove unused synchronization decorator

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
This commit is contained in:
Mark McLoughlin 2012-07-17 04:28:02 +01:00 committed by Jenkins
parent 362576f268
commit 4ab214a35f
7 changed files with 0 additions and 547 deletions

View File

@ -116,9 +116,6 @@ core_opts = [
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 = [

View File

@ -156,7 +156,6 @@ class Service(object):
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
@ -374,7 +373,6 @@ class WSGIService(object):
:returns: None
"""
utils.cleanup_file_locks()
rpc.register_opts(FLAGS)
if self.manager:
self.manager.init_host()

View File

@ -92,93 +92,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_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)

View File

@ -731,195 +731,6 @@ class DeprecationTest(test.TestCase):
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):

View File

@ -37,20 +37,17 @@ import socket
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
@ -66,10 +63,6 @@ ISO_TIME_FORMAT = "%Y-%m-%dT%H:%M:%S"
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.
@ -718,192 +711,6 @@ else:
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"""

View File

@ -95,8 +95,6 @@
# 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
@ -255,11 +253,6 @@
###### (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

View File

@ -1,63 +0,0 @@
#!/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()