92db8eea35
Remove a number of limitations from the external locks. - They can be nested now - They do not need external cleanup in case of failures - They do not rely on lockfile or greenlet internal implementation New implementation is based on fcntl locks and any crashing process will drop the lock. It does not have to rely on any cleanup code or handling exceptions. Because no cleanup is needed, a number of tests have been removed. This implementation is not portable outside of POSIX/BSD/SVR4 systems. Fcntl locks should work correctly with NFS mounts. Locks are cleaned up after the tests finish running via run_tests.sh, even though it's not strictly needed. This change requires eventlet >= 0.9.17. bp improve-external-locking Change-Id: Idf5424c04645f25097733848a007b150145b0b27
155 lines
5.0 KiB
Python
155 lines
5.0 KiB
Python
# vim: tabstop=4 shiftwidth=4 softtabstop=4
|
|
|
|
# Copyright 2010 OpenStack LLC
|
|
#
|
|
# 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 glob
|
|
import os
|
|
import select
|
|
|
|
from eventlet import greenpool
|
|
from eventlet import greenthread
|
|
|
|
from nova import exception
|
|
from nova import test
|
|
from nova import utils
|
|
|
|
|
|
class ExceptionTestCase(test.TestCase):
|
|
@staticmethod
|
|
def _raise_exc(exc):
|
|
raise exc()
|
|
|
|
def test_exceptions_raise(self):
|
|
for name in dir(exception):
|
|
exc = getattr(exception, name)
|
|
if isinstance(exc, type):
|
|
self.assertRaises(exc, self._raise_exc, exc)
|
|
|
|
|
|
class ProjectTestCase(test.TestCase):
|
|
|
|
def test_all_migrations_have_downgrade(self):
|
|
topdir = os.path.normpath(os.path.dirname(__file__) + '/../../')
|
|
py_glob = os.path.join(topdir, "nova", "db", "sqlalchemy",
|
|
"migrate_repo", "versions", "*.py")
|
|
missing_downgrade = []
|
|
for path in glob.iglob(py_glob):
|
|
has_upgrade = False
|
|
has_downgrade = False
|
|
with open(path, "r") as f:
|
|
for line in f:
|
|
if 'def upgrade(' in line:
|
|
has_upgrade = True
|
|
if 'def downgrade(' in line:
|
|
has_downgrade = True
|
|
|
|
if has_upgrade and not has_downgrade:
|
|
fname = os.path.basename(path)
|
|
missing_downgrade.append(fname)
|
|
|
|
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"""
|
|
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"""
|
|
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)
|