Skip to content

Z
ZODB
Commit 01f9261a authored by Jim Fulton's avatar Jim Fulton
Browse files
Options

Refactored cache verification to fix threading bugs during connection. 

Changed connections to work with unset (None) clients.  Messages
aren't forwarded until the client is set.  This is to prevent sending
spurious invalidation messages until a client is ready to recieve them.
parent 2dd88889
Show whitespace changes
Inline Side-by-side
Showing with 241 additions and 104 deletions
src/ZEO/ClientStorage.py
View file @ 01f9261a
......@@ -275,7 +275,7 @@ class ClientStorage(object):
# _server_addr is used by sortKey()
self._server_addr = None
self._verification_invalidations = None
self._pickler = self._tfile = None
self._info = {'length': 0, 'size': 0, 'name': 'ZEO Client',
'supportsUndo': 0, 'interfaces': ()}
......@@ -502,6 +502,7 @@ class ClientStorage(object):
# If we are upgrading from a read-only fallback connection,
# we must close the old connection to prevent it from being
# used while the cache is verified against the new connection.
self._connection.register_object(None) # Don't call me!
self._connection.close()
self._connection = None
self._ready.clear()
......@@ -576,60 +577,6 @@ class ClientStorage(object):
else:
return '%s:%s' % (self._storage, self._server_addr)
def verify_cache(self, server):
"""Internal routine called to verify the cache.
The return value (indicating which path we took) is used by
the test suite.
"""
# If verify_cache() finishes the cache verification process,
# it should set self._server. If it goes through full cache
# verification, then endVerify() should set self._server.
ltid = server.lastTransaction()
if not self._cache:
log2("No verification necessary -- empty cache")
self._server = server
if ltid and ltid != utils.z64:
self._cache.setLastTid(ltid)
self._ready.set()
return "full verification"
last_inval_tid = self._cache.getLastTid()
if last_inval_tid is not None:
if ltid == last_inval_tid:
log2("No verification necessary (last_inval_tid up-to-date)")
self._server = server
self._ready.set()
return "no verification"
# log some hints about last transaction
log2("last inval tid: %r %s\n"
% (last_inval_tid, tid2time(last_inval_tid)))
log2("last transaction: %r %s" %
(ltid, ltid and tid2time(ltid)))
pair = server.getInvalidations(last_inval_tid)
if pair is not None:
log2("Recovering %d invalidations" % len(pair[1]))
self.invalidateTransaction(*pair)
self._server = server
self._ready.set()
return "quick verification"
log2("Verifying cache")
# setup tempfile to hold zeoVerify results
self._verification_invalidations = []
# TODO: should batch these operations for efficiency; would need
# to acquire lock ...
for oid, tid in self._cache.contents():
server.verify(oid, tid)
self._pending_server = server
server.endZeoVerify()
return "full verification"
### Is there a race condition between notifyConnected and
### notifyDisconnected? In Particular, what if we get
### notifyDisconnected in the middle of notifyConnected?
......@@ -746,7 +693,7 @@ class ClientStorage(object):
try:
t = self._cache.load(oid)
if t:
return t[:2] # XXX strip version
return t
finally:
self._lock.release()
......@@ -1133,7 +1080,7 @@ class ClientStorage(object):
return
for oid, data in self._tbuf:
self._cache.invalidate(oid, tid)
self._cache.invalidate(oid, tid, False)
# If data is None, we just invalidate.
if data is not None:
s = self._seriald[oid]
......@@ -1218,6 +1165,63 @@ class ClientStorage(object):
"""Server callback to update the info dictionary."""
self._info.update(dict)
def verify_cache(self, server):
"""Internal routine called to verify the cache.
The return value (indicating which path we took) is used by
the test suite.
"""
self._pending_server = server
# setup tempfile to hold zeoVerify results and interim
# invalidation results
self._tfile = tempfile.TemporaryFile(suffix=".inv")
self._pickler = cPickle.Pickler(self._tfile, 1)
self._pickler.fast = 1 # Don't use the memo
# allow incoming invalidations:
self._connection.register_object(self)
# If verify_cache() finishes the cache verification process,
# it should set self._server. If it goes through full cache
# verification, then endVerify() should self._server.
ltid = server.lastTransaction()
if not self._cache:
log2("No verification necessary -- empty cache")
if ltid and ltid != utils.z64:
self._cache.setLastTid(ltid)
self.finish_verification()
return "full verification"
last_inval_tid = self._cache.getLastTid()
if last_inval_tid is not None:
if ltid == last_inval_tid:
log2("No verification necessary (last_inval_tid up-to-date)")
self.finish_verification()
return "no verification"
# log some hints about last transaction
log2("last inval tid: %r %s\n"
% (last_inval_tid, tid2time(last_inval_tid)))
log2("last transaction: %r %s" %
(ltid, ltid and tid2time(ltid)))
pair = server.getInvalidations(last_inval_tid)
if pair is not None:
log2("Recovering %d invalidations" % len(pair[1]))
self.finish_verification(pair)
return "quick verification"
elif ltid and ltid != utils.z64:
self._cache.setLastTid(ltid)
log2("Verifying cache")
for oid, tid in self._cache.contents():
server.verify(oid, tid)
server.endZeoVerify()
return "full verification"
def invalidateVerify(self, args):
"""Server callback to invalidate an (oid, '') pair.
......@@ -1225,53 +1229,78 @@ class ClientStorage(object):
"""
# Invalidation as result of verify_cache().
# Queue an invalidate for the end the verification procedure.
if self._verification_invalidations is None:
# This should never happen. TODO: assert it doesn't, or log
# if it does.
if self._pickler is None:
log2("invalidateVerify with no _pickler", level = logging.ERROR)
return
self._verification_invalidations.append(args[0])
self._pickler.dump((None, [args[0]]))
def _process_invalidations(self, tid, oids):
def endVerify(self):
"""Server callback to signal end of cache validation."""
log2("endVerify finishing")
self.finish_verification()
log2("endVerify finished")
def finish_verification(self, catch_up=None):
self._lock.acquire()
try:
for oid in oids:
if oid == self._load_oid:
self._load_status = 0
self._cache.invalidate(oid, tid)
if self._db is not None:
self._db.invalidate(tid, oids)
finally:
self._lock.release()
if catch_up:
# process catch-up invalidations
tid, invalidations = catch_up
self._process_invalidations(
tid, (arg[0] for arg in invalidations)
)
def endVerify(self):
"""Server callback to signal end of cache validation."""
if self._verification_invalidations is None:
if self._pickler is None:
return
# write end-of-data marker
self._process_invalidations(None, self._verification_invalidations)
self._verification_invalidations = None
self._pickler.dump((None, None))
self._pickler = None
self._tfile.seek(0)
unpickler = cPickle.Unpickler(self._tfile)
min_tid = self._cache.getLastTid()
while 1:
tid, oids = unpickler.load()
if oids is None:
break
if ((tid is None)
or (min_tid is None)
or (tid > min_tid)
):
self._process_invalidations(tid, oids)
self._tfile.close()
self._tfile = None
finally:
self._lock.release()
log2("endVerify finishing")
self._server = self._pending_server
self._ready.set()
self._pending_conn = None
log2("endVerify finished")
self._pending_server = None
def invalidateTransaction(self, tid, args):
"""Invalidate objects modified by tid."""
oids = (a[0] for a in args)
"""Server callback: Invalidate objects modified by tid."""
self._lock.acquire()
try:
self._cache.setLastTid(tid)
finally:
self._lock.release()
if self._verification_invalidations is not None:
if self._pickler is not None:
log2("Transactional invalidation during cache verification",
level=BLATHER)
self._verification_invalidations.extend(oids)
return
self._process_invalidations(tid, list(oids))
self._pickler.dump((tid, [arg[0] for arg in args]))
else:
self._process_invalidations(tid, (arg[0] for arg in args))
finally:
self._lock.release()
def _process_invalidations(self, tid, oids):
oids = list(oids)
for oid in oids:
if oid == self._load_oid:
self._load_status = 0
self._cache.invalidate(oid, tid)
if self._db is not None:
self._db.invalidate(tid, oids)
# The following are for compatibility with protocol version 2.0.0
......
src/ZEO/tests/invalidations_while_connecting.test 0 → 100644
View file @ 01f9261a
Invalidations while connecting
==============================
As soon as a client registers with a server, it will recieve
invalidations from the server. The client must be careful to queue
these invalidations until it is ready to deal with them. At the time
of the writing of this test, clients weren't careful enogh about
queing invalidations. This led to cache corruption in the form of
both low-level file corruption as well as out-of-date records marked
as current.
This tests tries to provoke this bug by:
- starting a server
>>> import ZEO.tests.testZEO, ZEO.tests.forker
>>> addr = 'localhost', ZEO.tests.testZEO.get_port()
>>> zconf = ZEO.tests.forker.ZEOConfig(addr)
>>> sconf = '<filestorage 1>\npath Data.fs\n</filestorage>\n'
>>> _, adminaddr, pid, conf_path = ZEO.tests.forker.start_zeo_server(
... sconf, zconf, addr[1])
- opening a client to the server that writes some objects, filling
it's cache at the same time,
>>> import ZEO.ClientStorage, ZODB.tests.MinPO, transaction
>>> db = ZODB.DB(ZEO.ClientStorage.ClientStorage(addr, client='x'))
>>> conn = db.open()
>>> nobs = 1000
>>> for i in range(nobs):
... conn.root()[i] = ZODB.tests.MinPO.MinPO(0)
>>> transaction.commit()
- disconnecting the first client (closing it with a persistent cache),
>>> db.close()
- starting a second client that writes objects more or less
constantly,
>>> import random, threading
>>> stop = False
>>> db2 = ZODB.DB(ZEO.ClientStorage.ClientStorage(addr))
>>> tm = transaction.TransactionManager()
>>> conn2 = db2.open(transaction_manager=tm)
>>> random = random.Random(0)
>>> lock = threading.Lock()
>>> def run():
... while 1:
... i = random.randint(0, nobs-1)
... if stop:
... return
... lock.acquire()
... try:
... conn2.root()[i].value += 1
... tm.commit()
... finally:
... lock.release()
... time.sleep(0)
>>> thread = threading.Thread(target=run)
>>> thread.start()
- restarting the first client, and
- testing for cache validity.
>>> import zope.testing.loggingsupport, logging
>>> handler = zope.testing.loggingsupport.InstalledHandler(
... 'ZEO', level=logging.ERROR)
>>> import time
>>> for c in range(10):
... time.sleep(.1)
... db = ZODB.DB(ZEO.ClientStorage.ClientStorage(addr, client='x'))
... _ = lock.acquire()
... try:
... time.sleep(.1)
... assert (db._storage.lastTransaction()
... == db._storage._server.lastTransaction()), (
... db._storage.lastTransaction(),
... db._storage._server.lastTransactiion())
... conn = db.open()
... for i in range(1000):
... if conn.root()[i].value != conn2.root()[i].value:
... print 'bad', c, i, conn.root()[i].value,
... print conn2.root()[i].value
... finally:
... _ = lock.release()
... db.close()
>>> stop = True
>>> thread.join(10)
>>> thread.isAlive()
False
>>> for record in handler.records:
... print record.name, record.levelname
... print handler.format(record)
>>> handler.uninstall()
>>> db.close()
>>> db2.close()
src/ZEO/tests/testConnection.py
View file @ 01f9261a
......@@ -21,7 +21,7 @@ platform-dependent scaffolding.
import unittest
# Import the actual test class
from ZEO.tests import ConnectionTests, InvalidationTests
from zope.testing import doctest, setupstack
class FileStorageConfig:
def getConfig(self, path, create, read_only):
......@@ -135,6 +135,10 @@ def test_suite():
for klass in test_classes:
sub = unittest.makeSuite(klass, 'check')
suite.addTest(sub)
suite.addTest(doctest.DocFileSuite(
'invalidations_while_connecting.test',
setUp=setupstack.setUpDirectory, tearDown=setupstack.tearDown,
))
return suite
......
src/ZEO/zrpc/client.py
View file @ 01f9261a
......@@ -447,8 +447,7 @@ class ConnectWrapper:
Call the client's testConnection(), giving the client a chance
to do app-level check of the connection.
"""
self.conn = ManagedClientConnection(self.sock, self.addr,
self.client, self.mgr)
self.conn = ManagedClientConnection(self.sock, self.addr, self.mgr)
self.sock = None # The socket is now owned by the connection
try:
self.preferred = self.client.testConnection(self.conn)
......
src/ZEO/zrpc/connection.py
View file @ 01f9261a
......@@ -544,14 +544,23 @@ class Connection(smac.SizedMessageAsyncConnection, object):
self.replies_cond.release()
def handle_request(self, msgid, flags, name, args):
if not self.check_method(name):
msg = "Invalid method name: %s on %s" % (name, repr(self.obj))
obj = self.obj
if name.startswith('_') or not hasattr(obj, name):
if obj is None:
if __debug__:
self.log("no object calling %s%s"
% (name, short_repr(args)),
level=logging.DEBUG)
return
msg = "Invalid method name: %s on %s" % (name, repr(obj))
raise ZRPCError(msg)
if __debug__:
self.log("calling %s%s" % (name, short_repr(args)),
level=logging.DEBUG)
meth = getattr(self.obj, name)
meth = getattr(obj, name)
try:
self.waiting_for_reply = True
try:
......@@ -590,12 +599,6 @@ class Connection(smac.SizedMessageAsyncConnection, object):
level=logging.ERROR, exc_info=True)
self.close()
def check_method(self, name):
# TODO: This is hardly "secure".
if name.startswith('_'):
return None
return hasattr(self.obj, name)
def send_reply(self, msgid, ret):
# encode() can pass on a wide variety of exceptions from cPickle.
# While a bare `except` is generally poor practice, in this case
......@@ -813,7 +816,7 @@ class ManagedClientConnection(Connection):
__super_close = Connection.close
base_message_output = Connection.message_output
def __init__(self, sock, addr, obj, mgr):
def __init__(self, sock, addr, mgr):
self.mgr = mgr
# We can't use the base smac's message_output directly because the
......@@ -830,7 +833,7 @@ class ManagedClientConnection(Connection):
self.queue_output = True
self.queued_messages = []
self.__super_init(sock, addr, obj, tag='C', map=client_map)
self.__super_init(sock, addr, None, tag='C', map=client_map)
self.trigger = client_trigger
client_trigger.pull_trigger()
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment
GitLab Nexedi Edition | About GitLab | About Nexedi | 沪ICP备2021021310号-2 | 沪ICP备2021021310号-7