Commit f4f8a7ab authored by Jim Fulton's avatar Jim Fulton

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 84162584
This diff is collapsed.
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()
......@@ -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
......
......@@ -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)
......
......@@ -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