Skip to content

Z
ZEO
Commit 3f31236b authored by Jim Fulton's avatar Jim Fulton
Browse files
Options

Checkpointing.... 

Many tests passing. Quite a few still failing.
parent bdbc36dd
Show whitespace changes
Inline Side-by-side
Showing with 391 additions and 1649 deletions
setup.py
View file @ 3f31236b
......@@ -118,7 +118,7 @@ setup(name="ZEO",
install_requires = [
'ZODB >= 4.2.0b1',
'six',
'transaction',
'transaction >= 1.6.0',
'persistent >= 4.1.0',
'zc.lockfile',
'ZConfig',
......
src/ZEO/ClientStorage.py
View file @ 3f31236b
......@@ -18,39 +18,34 @@ Public contents of this module:
ClientStorage -- the main class, implementing the Storage API
"""
import BTrees.IOBTree
import gc
import logging
import os
import re
import socket
import stat
import sys
import tempfile
import threading
import time
import weakref
from binascii import hexlify
import zc.lockfile
import ZEO.interfaces
import ZODB
import ZODB.BaseStorage
import ZODB.interfaces
import ZODB.event
import zope.interface
import six
from persistent.TimeStamp import TimeStamp
from ZEO._compat import Pickler, Unpickler, get_ident, PY3
from ZEO.auth import get_module
from ZEO.cache import ClientCache
from ZEO.Exceptions import ClientStorageError, ClientDisconnected, AuthError
from ZEO import ServerStub
from ZEO._compat import get_ident
from ZEO.Exceptions import ClientDisconnected
from ZEO.TransactionBuffer import TransactionBuffer
from ZEO.zrpc.client import ConnectionManager
from ZODB import POSException
from ZODB import utils
import ZEO.asyncio.client
import ZEO.cache
logger = logging.getLogger(__name__)
try:
......@@ -75,22 +70,6 @@ def get_timestamp(prev_ts=None):
t = t.laterThan(prev_ts)
return t
class DisconnectedServerStub:
"""Internal helper class used as a faux RPC stub when disconnected.
This raises ClientDisconnected on all attribute accesses.
This is a singleton class -- there should be only one instance,
the global disconnected_stub, so it can be tested by identity.
"""
def __getattr__(self, attr):
raise ClientDisconnected()
# Singleton instance of DisconnectedServerStub
disconnected_stub = DisconnectedServerStub()
MB = 1024**2
class ClientStorage(object):
......@@ -103,28 +82,20 @@ class ClientStorage(object):
"""
# ClientStorage does not declare any interfaces here. Interfaces are
# declared according to the server's storage once a connection is
# established.
# Classes we instantiate. A subclass might override.
TransactionBufferClass = TransactionBuffer
ClientCacheClass = ClientCache
ConnectionManagerClass = ConnectionManager
StorageServerStubClass = ServerStub.stub
def __init__(self, addr, storage='1', cache_size=20 * MB,
name='', client=None, var=None,
min_disconnect_poll=1, max_disconnect_poll=30,
wait_for_server_on_startup=None, # deprecated alias for wait
wait=None, wait_timeout=None,
name='', wait_timeout=None,
disconnect_poll=None,
read_only=0, read_only_fallback=0,
drop_cache_rather_verify=False,
username='', password='', realm=None,
blob_dir=None, shared_blob_dir=False,
blob_cache_size=None, blob_cache_size_check=10,
client_label=None,
cache=None,
# Mostly ignored backward-compatability options
client=None, var=None,
min_disconnect_poll=1, max_disconnect_poll=None,
wait=True,
drop_cache_rather_verify=True,
username=None, password=None, realm=None,
):
"""ClientStorage constructor.
......@@ -142,50 +113,31 @@ class ClientStorage(object):
address. Required.
storage
The storage name, defaulting to '1'. The name must
The server storage name, defaulting to '1'. The name must
match one of the storage names supported by the server(s)
specified by the addr argument. The storage name is
displayed in the Zope control panel.
specified by the addr argument.
cache_size
The disk cache size, defaulting to 20 megabytes.
This is passed to the ClientCache constructor.
name
The storage name, defaulting to ''. If this is false,
str(addr) is used as the storage name.
client
A name used to construct persistent cache filenames.
Defaults to None, in which case the cache is not persistent.
See ClientCache for more info.
var
When client is not None, this specifies the directory
where the persistent cache files are created. It defaults
to None, in which case the current directory is used.
min_disconnect_poll
The minimum delay in seconds between
attempts to connect to the server, in seconds. Defaults
to 1 second.
The storage name, defaulting to a combination of the
address and the server storage name. This is used to
construct the response to getName()
max_disconnect_poll
The maximum delay in seconds between
attempts to connect to the server, in seconds. Defaults
to 300 seconds.
cache
A cache object or a name, relative to the current working
directory, used to construct persistent cache filenames.
Defaults to None, in which case the cache is not
persistent. See ClientCache for more info.
wait_for_server_on_startup
A backwards compatible alias for
the wait argument.
wait
A flag indicating whether to wait until a connection
with a server is made, defaulting to true.
disconnect_poll
The delay in seconds between attempts to connect to the
server, in seconds. Defaults to 1 second.
wait_timeout
Maximum time to wait for a connection before
giving up. Only meaningful if wait is True.
Maximum time to wait for results, including connecting.
read_only
A flag indicating whether this should be a
......@@ -199,21 +151,6 @@ class ClientStorage(object):
most one of read_only and read_only_fallback should be
true.
username
string with username to be used when authenticating.
These only need to be provided if you are connecting to an
authenticated server storage.
password
string with plaintext password to be used when authenticated.
realm
not documented.
drop_cache_rather_verify
a flag indicating that the cache should be dropped rather
than expensively verified.
blob_dir
directory path for blob data. 'blob data' is data that
is retrieved via the loadBlob API.
......@@ -246,10 +183,16 @@ class ClientStorage(object):
"""
if isinstance(addr, int):
addr = '127.0.0.1', addr
addr = ('127.0.0.1', addr)
self.__name__ = name or str(addr) # Standard convention for storages
if isinstance(addr, str):
addr = [addr]
elif (isinstance(addr, tuple) and len(addr) == 2 and
isinstance(addr[0], str) and isinstance(addr[1], int)):
addr = [addr]
logger.info(
"%s %s (pid=%d) created %s/%s for storage: %r",
self.__name__,
......@@ -260,121 +203,27 @@ class ClientStorage(object):
storage,
)
self._drop_cache_rather_verify = drop_cache_rather_verify
# wait defaults to True, but wait_for_server_on_startup overrides
# if not None
if wait_for_server_on_startup is not None:
if wait is not None and wait != wait_for_server_on_startup:
logger.warning(
"%s ClientStorage(): conflicting values for wait and "
"wait_for_server_on_startup; wait prevails",
self.__name__)
else:
logger.info(
"%s ClientStorage(): wait_for_server_on_startup "
"is deprecated; please use wait instead",
self.__name__)
wait = wait_for_server_on_startup
elif wait is None:
wait = 1
self._is_read_only = read_only
self._addr = addr # For tests
# A ZEO client can run in disconnected mode, using data from
# its cache, or in connected mode. Several instance variables
# are related to whether the client is connected.
# _server: All method calls are invoked through the server
# stub. When not connect, set to disconnected_stub an
# object that raises ClientDisconnected errors.
# _ready: A threading Event that is set only if _server
# is set to a real stub.
# _connection: The current zrpc connection or None.
# _connection is set as soon as a connection is established,
# but _server is set only after cache verification has finished
# and clients can safely use the server. _pending_server holds
# a server stub while it is being verified.
self._server = disconnected_stub
self._connection = None
self._pending_server = None
self._ready = threading.Event()
# _is_read_only stores the constructor argument
self._is_read_only = read_only
self._storage = storage
self._read_only_fallback = read_only_fallback
self._username = username
self._password = password
self._realm = realm
self._iterators = weakref.WeakValueDictionary()
self._iterator_ids = set()
# Flag tracking disconnections in the middle of a transaction. This
# is reset in tpc_begin() and set in notifyDisconnected().
self._midtxn_disconnect = 0
self._storage = storage
# _server_addr is used by sortKey()
self._server_addr = None
self._client_label = client_label
self._pickler = self._tfile = None
self._info = {'length': 0, 'size': 0, 'name': 'ZEO Client',
'supportsUndo': 0, 'interfaces': ()}
self._tbuf = self.TransactionBufferClass()
self._db = None
self._ltid = None # the last committed transaction
# _serials: stores (oid, serialno) as returned by server
# _seriald: _check_serials() moves from _serials to _seriald,
# which maps oid to serialno
# TODO: If serial number matches transaction id, then there is
# no need to have all this extra infrastructure for handling
# serial numbers. The vote call can just return the tid.
# If there is a conflict error, we can't have a special method
# called just to propagate the error.
self._serials = []
self._seriald = {}
# A ClientStorage only allows one thread to commit at a time.
# Mutual exclusion is achieved using _tpc_cond, which
# protects _transaction. A thread that wants to assign to
# self._transaction must acquire _tpc_cond first. A thread
# that decides it's done with a transaction (whether via success
# or failure) must set _transaction to None and do
# _tpc_cond.notify() before releasing _tpc_cond.
self._tpc_cond = threading.Condition()
self._transaction = None
# Prevent multiple new_oid calls from going out. The _oids
# variable should only be modified while holding the
# _oid_lock.
self._oid_lock = threading.Lock()
self._oids = [] # Object ids retrieved from new_oids()
# load() and tpc_finish() must be serialized to guarantee
# that cache modifications from each occur atomically.
# It also prevents multiple load calls occuring simultaneously,
# which simplifies the cache logic.
self._load_lock = threading.Lock()
# _load_oid and _load_status are protected by _lock
self._load_oid = None
self._load_status = None
# Can't read data in one thread while writing data
# (tpc_finish) in another thread. In general, the lock
# must prevent access to the cache while _update_cache
# is executing.
self._lock = threading.Lock()
self._oids = [] # List of pre-fetched oids from server
cache = self._cache = open_cache(cache, var, client, cache_size)
# XXX need to check for POSIX-ness here
self.blob_dir = blob_dir
......@@ -396,15 +245,6 @@ class ClientStorage(object):
else:
self.fshelper = None
if client is not None:
dir = var or os.getcwd()
cache_path = os.path.join(dir, "%s-%s.zec" % (client, storage))
else:
cache_path = None
self._cache = self.ClientCacheClass(cache_path, size=cache_size)
self._blob_cache_size = blob_cache_size
self._blob_data_bytes_loaded = 0
if blob_cache_size is not None:
......@@ -413,65 +253,25 @@ class ClientStorage(object):
blob_cache_size * blob_cache_size_check // 100)
self._check_blob_size()
self._rpc_mgr = self.ConnectionManagerClass(addr, self,
tmin=min_disconnect_poll,
tmax=max_disconnect_poll)
self._server = ZEO.asyncio.client.ClientThread(
addr, self, cache, storage,
ZEO.asyncio.client.Fallback if read_only_fallback else read_only,
wait_timeout or 30,
wait=wait,
)
self._call = self._server.call
self._async = self._server.async
self._async_iter = self._server.async_iter
if wait:
self._wait(wait_timeout)
else:
# attempt_connect() will make an attempt that doesn't block
# "too long," for a very vague notion of too long. If that
# doesn't succeed, call connect() to start a thread.
if not self._rpc_mgr.attempt_connect():
self._rpc_mgr.connect()
self._commit_lock = threading.Lock()
def new_addr(self, addr):
self._addr = addr
self._rpc_mgr.new_addrs(addr)
def _wait(self, timeout=None):
if timeout is not None:
deadline = time.time() + timeout
logger.debug("%s Setting deadline to %f", self.__name__, deadline)
else:
deadline = None
# Wait for a connection to be established.
self._rpc_mgr.connect(sync=1)
# When a synchronous connect() call returns, there is
# a valid _connection object but cache validation may
# still be going on. This code must wait until validation
# finishes, but if the connection isn't a zrpc async
# connection it also needs to poll for input.
while 1:
self._ready.wait(30)
if self._ready.isSet():
break
if timeout and time.time() > deadline:
logger.warning("%s Timed out waiting for connection",
self.__name__)
break
logger.info("%s Waiting for cache verification to finish",
self.__name__)
self._server.new_addrs(addr)
def close(self):
"Storage API: finalize the storage, releasing external resources."
_rpc_mgr = self._rpc_mgr
self._rpc_mgr = None
if _rpc_mgr is None:
return # already closed
if self._connection is not None:
self._connection.register_object(None) # Don't call me!
self._connection = None
_rpc_mgr.close()
self._tbuf.close()
if self._cache is not None:
self._cache.close()
self._cache = None
if self._tfile is not None:
self._tfile.close()
self._server.close()
if self._check_blob_size_thread is not None:
self._check_blob_size_thread.join()
......@@ -510,153 +310,43 @@ class ClientStorage(object):
def is_connected(self, test=False):
"""Return whether the storage is currently connected to a server."""
# This function is used by clients, so we only report that a
# connection exists when the connection is ready to use.
if test:
try:
self._server.lastTransaction()
except Exception:
pass
return self._ready.isSet()
return self._server.is_connected()
def sync(self):
# The separate async thread should keep us up to date
pass
def doAuth(self, protocol, stub):
if not (self._username and self._password):
raise AuthError("empty username or password")
module = get_module(protocol)
if not module:
logger.error("%s %s: no such an auth protocol: %s",
self.__name__, self.__class__.__name__, protocol)
return
storage_class, client, db_class = module
if not client:
logger.error(
"%s %s: %s isn't a valid protocol, must have a Client class",
self.__name__, self.__class__.__name__, protocol)
raise AuthError("invalid protocol")
c = client(stub)
# Initiate authentication, returns boolean specifying whether OK
return c.start(self._username, self._realm, self._password)
def testConnection(self, conn):
"""Internal: test the given connection.
This returns: 1 if the connection is an optimal match,
0 if it is a suboptimal but acceptable match.
It can also raise DisconnectedError or ReadOnlyError.
This is called by ZEO.zrpc.ConnectionManager to decide which
connection to use in case there are multiple, and some are
read-only and others are read-write.
This works by calling register() on the server. In read-only
mode, register() is called with the read_only flag set. In
writable mode and in read-only fallback mode, register() is
called with the read_only flag cleared. In read-only fallback
mode only, if the register() call raises ReadOnlyError, it is
retried with the read-only flag set, and if this succeeds,
this is deemed a suboptimal match. In all other cases, a
succeeding register() call is deemed an optimal match, and any
exception raised by register() is passed through.
"""
logger.info("%s Testing connection %r", self.__name__, conn)
# TODO: Should we check the protocol version here?
conn._is_read_only = self._is_read_only
stub = self.StorageServerStubClass(conn)
auth = stub.getAuthProtocol()
logger.info("%s Server authentication protocol %r", self.__name__, auth)
if auth:
skey = self.doAuth(auth, stub)
if skey:
logger.info("%s Client authentication successful",
self.__name__)
conn.setSessionKey(skey)
else:
logger.info("%s Authentication failed",
self.__name__)
raise AuthError("Authentication failed")
try:
stub.register(str(self._storage), self._is_read_only)
return 1
except POSException.ReadOnlyError:
if not self._read_only_fallback:
raise
logger.info("%s Got ReadOnlyError; trying again with read_only=1",
self.__name__)
stub.register(str(self._storage), read_only=1)
conn._is_read_only = True
return 0
def notifyConnected(self, conn):
"""Internal: start using the given connection.
This is called by ConnectionManager after it has decided which
connection should be used.
"""
if self._cache is None:
# the storage was closed, but the connect thread called
# this method before it was stopped.
return
if self._connection is not None:
# 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()
reconnect = 1
else:
reconnect = 0
self.set_server_addr(conn.get_addr())
self._connection = conn
_connection_generation = 0
def notify_connected(self, conn, info):
reconnected = self._connection_generation
self.set_server_addr(conn.get_peername())
self.protocol_version = conn.protocol_version
self._is_read_only = conn.is_read_only()
# invalidate our db cache
if self._db is not None:
self._db.invalidateCache()
if reconnect:
logger.info("%s Reconnected to storage: %s",
self.__name__, self._server_addr)
else:
logger.info("%s Connected to storage: %s",
self.__name__, self._server_addr)
stub = self.StorageServerStubClass(conn)
if self._client_label and conn.peer_protocol_version >= b"Z310":
stub.set_client_label(self._client_label)
logger.info("%s %s to storage: %s",
self.__name__,
'Reconnected' if self._connection_generation
else 'Connected',
self._server_addr)
if conn.peer_protocol_version < b"Z3101":
logger.warning("Old server doesn't suppport "
"checkCurrentSerialInTransaction")
self.checkCurrentSerialInTransaction = lambda *args: None
self._connection_generation += 1
self._oids = []
self.verify_cache(stub)
if self._client_label:
conn.call_async_from_same_thread(
'set_client_label', self._client_label)
# It's important to call get_info after calling verify_cache.
# If we end up doing a full-verification, we need to wait till
# it's done. By doing a synchonous call, we are guarenteed
# that the verification will be done because operations are
# handled in order.
self._info.update(stub.get_info())
self._info.update(info)
self._handle_extensions()
# for name in self._info.get('extensionMethods', {}).keys():
# if not hasattr(self, name):
# def mklambda(mname):
# return (lambda *args, **kw:
# self._server.rpc.call(mname, *args, **kw))
# setattr(self, name, mklambda(name))
for iface in (
ZODB.interfaces.IStorageRestoreable,
......@@ -670,14 +360,6 @@ class ClientStorage(object):
'interfaces', ()):
zope.interface.alsoProvides(self, iface)
def _handle_extensions(self):
for name in self.getExtensionMethods().keys():
if not hasattr(self, name):
def mklambda(mname):
return (lambda *args, **kw:
self._server.rpc.call(mname, *args, **kw))
setattr(self, name, mklambda(name))
def set_server_addr(self, addr):
# Normalize server address and convert to string
if isinstance(addr, str):
......@@ -699,6 +381,8 @@ class ClientStorage(object):
self._server_addr = str((canonical, addr[1]))
def sortKey(self):
# XXX sortKey should be explicit, possibly based on database name.
# If the client isn't connected to anything, it can't have a
# valid sortKey(). Raise an error to stop the transaction early.
if self._server_addr is None:
......@@ -706,15 +390,7 @@ class ClientStorage(object):
else:
return '%s:%s' % (self._storage, self._server_addr)
### Is there a race condition between notifyConnected and
### notifyDisconnected? In Particular, what if we get
### notifyDisconnected in the middle of notifyConnected?
### The danger is that we'll proceed as if we were connected
### without worrying if we were, but this would happen any way if
### notifyDisconnected had to get the instance lock. There's
### nothing to gain by getting the instance lock.
def notifyDisconnected(self):
def notify_disconnected(self):
"""Internal: notify that the server connection was terminated.
This is called by ConnectionManager when the connection is
......@@ -723,11 +399,9 @@ class ClientStorage(object):
"""
logger.info("%s Disconnected from storage: %r",
self.__name__, self._server_addr)
self._connection = None
self._ready.clear()
self._server = disconnected_stub
self._midtxn_disconnect = 1
self._iterator_gc(True)
self._connection_generation += 1
self._is_read_only = self._server.is_read_only()
def __len__(self):
"""Return the size of the storage."""
......@@ -752,156 +426,73 @@ class ClientStorage(object):
"""Storage API: an approximate size of the database, in bytes."""
return self._info['size']
def getExtensionMethods(self):
"""getExtensionMethods
This returns a dictionary whose keys are names of extra methods
provided by this storage. Storage proxies (such as ZEO) should
call this method to determine the extra methods that they need
to proxy in addition to the standard storage methods.
Dictionary values should be None; this will be a handy place
for extra marshalling information, should we need it
"""
return self._info.get('extensionMethods', {})
def supportsUndo(self):
"""Storage API: return whether we support undo."""
return self._info['supportsUndo']
def isReadOnly(self):
"""Storage API: return whether we are in read-only mode."""
if self._is_read_only:
return True
else:
# If the client is configured for a read-write connection
# but has a read-only fallback connection, conn._is_read_only
# will be True. If self._connection is None, we'll behave as
# read_only
try:
return self._connection._is_read_only
except AttributeError:
return True
def is_read_only(self):
"""Storage API: return whether we are in read-only mode.
"""
return self._is_read_only or self._server.is_read_only()
isReadOnly = is_read_only
def _check_trans(self, trans):
def _check_trans(self, trans, meth):
"""Internal helper to check a transaction argument for sanity."""
if self._is_read_only:
raise POSException.ReadOnlyError()
if self._transaction is not trans:
raise POSException.StorageTransactionError(self._transaction,
trans)
try:
buf = trans.data(self)
except KeyError:
raise POSException.StorageTransactionError(
"Transaction not committing", meth, trans)
if buf.connection_generation != self._connection_generation:
# We were disconneected, so this one is poisoned
raise ClientDisconnected(meth, 'on a disconnected transaction')
return buf
def history(self, oid, size=1):
"""Storage API: return a sequence of HistoryEntry objects.
"""
return self._server.history(oid, size)
return self._call('history', oid, size)
def record_iternext(self, next=None):
"""Storage API: get the next database record.
This is part of the conversion-support API.
"""
return self._server.record_iternext(next)
return self._call('record_iternext', next)
def getTid(self, oid):
"""Storage API: return current serial number for oid."""
return self._server.getTid(oid)
# XXX deprecated: used by storage server for full cache verification.
return self._call('getTid', oid)
def loadSerial(self, oid, serial):
"""Storage API: load a historical revision of an object."""
return self._server.loadSerial(oid, serial)
return self._call('loadSerial', oid, serial)
def load(self, oid, version=''):
"""Storage API: return the data for a given object.
This returns the pickle data and serial number for the object
specified by the given object id, if they exist;
otherwise a KeyError is raised.
"""
self._lock.acquire() # for atomic processing of invalidations
try:
t = self._cache.load(oid)
if t:
return t
finally:
self._lock.release()
if self._server is None:
raise ClientDisconnected()
self._load_lock.acquire()
try:
self._lock.acquire()
try:
self._load_oid = oid
self._load_status = 1
finally:
self._lock.release()
data, tid = self._server.loadEx(oid)
self._lock.acquire() # for atomic processing of invalidations
try:
if self._load_status:
self._cache.store(oid, tid, None, data)
self._load_oid = None
finally:
self._lock.release()
finally:
self._load_lock.release()
return data, tid
return self._server.load(oid)
def loadBefore(self, oid, tid):
self._lock.acquire()
try:
t = self._cache.loadBefore(oid, tid)
if t is not None:
return t
finally:
self._lock.release()
t = self._server.loadBefore(oid, tid)
if t is None:
return None
data, start, end = t
if end is None:
# This method should not be used to get current data. It
# doesn't use the _load_lock, so it is possble to overlap
# this load with an invalidation for the same object.
# If we call again, we're guaranteed to get the
# post-invalidation data. But if the data is still
# current, we'll still get end == None.
# Maybe the best thing to do is to re-run the test with
# the load lock in the case. That's slow performance, but
# I don't think real application code will ever care about
# it.
return data, start, end
self._lock.acquire()
try:
self._cache.store(oid, start, end, data)
finally:
self._lock.release()
return data, start, end
return self._server.load_before(oid, tid)
def new_oid(self):
"""Storage API: return a new object identifier."""
"""Storage API: return a new object identifier.
"""
if self._is_read_only:
raise POSException.ReadOnlyError()
# avoid multiple oid requests to server at the same time
self._oid_lock.acquire()
while 1:
try:
if not self._oids:
self._oids = self._server.new_oids()
self._oids.reverse()
return self._oids.pop()
finally:
self._oid_lock.release()
except IndexError:
pass # We ran out. We need to get some more.
self._oids[:0] = reversed(self._call('new_oids'))
def pack(self, t=None, referencesf=None, wait=1, days=0):
"""Storage API: pack the storage.
......@@ -922,40 +513,25 @@ class ClientStorage(object):
if t is None:
t = time.time()
t = t - (days * 86400)
return self._server.pack(t, wait)
def _check_serials(self):
"""Internal helper to move data from _serials to _seriald."""
# serials are always going to be the same, the only
# question is whether an exception has been raised.
if self._serials:
l = len(self._serials)
r = self._serials[:l]
del self._serials[:l]
for oid, s in r:
if isinstance(s, Exception):
self._cache.invalidate(oid, None)
raise s
self._seriald[oid] = s
return r
return self._call('pack', t, wait)
def store(self, oid, serial, data, version, txn):
"""Storage API: store data for an object."""
assert not version
self._check_trans(txn)
self._server.storea(oid, serial, data, id(txn))
self._tbuf.store(oid, data)
return self._check_serials()
tbuf = self._check_trans(txn, 'store')
self._async('storea', oid, serial, data, id(txn))
tbuf.store(oid, data)
def checkCurrentSerialInTransaction(self, oid, serial, transaction):
self._check_trans(transaction)
self._server.checkCurrentSerialInTransaction(oid, serial,
id(transaction))
self._check_trans(transaction, 'checkCurrentSerialInTransaction')
self._async(
'checkCurrentSerialInTransaction', oid, serial, id(transaction))
def storeBlob(self, oid, serial, data, blobfilename, version, txn):
"""Storage API: store a blob object."""
assert not version
tbuf = self._check_trans(txn, 'storeBlob')
# Grab the file right away. That way, if we don't have enough
# room for a copy, we'll know now rather than in tpc_finish.
......@@ -973,11 +549,29 @@ class ClientStorage(object):
serials = self.store(oid, serial, data, '', txn)
if self.shared_blob_dir:
self._server.storeBlobShared(
self._async(
'storeBlobShared',
oid, serial, data, os.path.basename(target), id(txn))
else:
self._server.storeBlob(oid, serial, data, target, txn)
self._tbuf.storeBlob(oid, target)
# Store a blob to the server. We don't want to real all of
# the data into memory, so we use a message iterator. This
# allows us to read the blob data as needed.
def store():
yield ('storeBlobStart', ())
f = open(blobfilename, 'rb')
while 1:
chunk = f.read(59000)
if not chunk:
break
yield ('storeBlobChunk', (chunk, ))
f.close()
yield ('storeBlobEnd', (oid, serial, data, id(txn)))
self._async_iter(store())
tbuf.storeBlob(oid, target)
return serials
def receiveBlobStart(self, oid, serial):
......@@ -1006,9 +600,9 @@ class ClientStorage(object):
os.chmod(blob_filename, stat.S_IREAD)
def deleteObject(self, oid, serial, txn):
self._check_trans(txn)
self._server.deleteObject(oid, serial, id(txn))
self._tbuf.store(oid, None)
tbuf = self._check_trans(txn, 'deleteObject')
self._async('deleteObject', oid, serial, id(txn))
tbuf.store(oid, None)
def loadBlob(self, oid, serial):
# Load a blob. If it isn't present and we have a shared blob
......@@ -1053,7 +647,7 @@ class ClientStorage(object):
# returns, it will have been sent. (The recieving will
# have been handled by the asyncore thread.)
self._server.sendBlob(oid, serial)
self._call('sendBlob', oid, serial)
if os.path.exists(blob_filename):
return _accessed(blob_filename)
......@@ -1083,7 +677,7 @@ class ClientStorage(object):
# We're using a server shared cache. If the file isn't
# here, it's not anywhere.
raise POSException.POSKeyError("No blob file", oid, serial)
self._server.sendBlob(oid, serial)
self._call('sendBlob', oid, serial)
if not os.path.exists(blob_filename):
raise POSException.POSKeyError("No blob file", oid, serial)
......@@ -1100,12 +694,15 @@ class ClientStorage(object):
return self.fshelper.temp_dir
def tpc_vote(self, txn):
"""Storage API: vote on a transaction."""
if txn is not self._transaction:
raise POSException.StorageTransactionError(
"tpc_vote called with wrong transaction")
self._server.vote(id(txn))
return self._check_serials()
"""Storage API: vote on a transaction.
"""
tbuf = self._check_trans(txn, 'tpc_vote')
self._call('vote', id(txn))
if tbuf.exception:
raise tbuf.exception
return list(tbuf.serials.items())
def tpc_transaction(self):
return self._transaction
......@@ -1114,136 +711,86 @@ class ClientStorage(object):
"""Storage API: begin a transaction."""
if self._is_read_only:
raise POSException.ReadOnlyError()
self._tpc_cond.acquire()
try:
self._midtxn_disconnect = 0
while self._transaction is not None:
# It is allowable for a client to call two tpc_begins in a
# row with the same transaction, and the second of these
# must be ignored.
if self._transaction == txn:
tbuf = txn.data(self)
except KeyError:
pass
else:
if tbuf is not None:
raise POSException.StorageTransactionError(
"Duplicate tpc_begin calls for same transaction")
self._tpc_cond.wait(30)
self._transaction = txn
finally:
self._tpc_cond.release()
txn.set_data(self, TransactionBuffer(self._connection_generation))
if self.protocol_version < b'Z5':
# Earlier protocols only allowed one transaction at a time :(
self._commit_lock.acquire()
self._tbuf = txn.data(self)
try:
self._server.tpc_begin(id(txn), txn.user, txn.description,
txn._extension, tid, status)
except:
# Client may have disconnected during the tpc_begin().
if self._server is not disconnected_stub:
self.end_transaction()
self._async(
'tpc_begin', id(txn),
txn.user, txn.description, txn._extension, tid, status)
except ClientDisconnected:
self.tpc_end(txn)
raise
self._tbuf.clear()
self._seriald.clear()
del self._serials[:]
def end_transaction(self):
"""Internal helper to end a transaction."""
# the right way to set self._transaction to None
# calls notify() on _tpc_cond in case there are waiting threads
self._tpc_cond.acquire()
try:
self._transaction = None
self._tpc_cond.notify()
finally:
self._tpc_cond.release()
def tpc_end(self, txn):
tbuf = txn.data(self)
if tbuf is not None:
tbuf.close()
txn.set_data(self, None)
if self.protocol_version < b'Z5':
self._commit_lock.release()
def lastTransaction(self):
return self._cache.getLastTid()
def tpc_abort(self, txn):
"""Storage API: abort a transaction."""
if txn is not self._transaction:
"""Storage API: abort a transaction.
"""
try:
tbuf = txn.data(self)
except KeyError:
return
try:
# Caution: Are there any exceptions that should prevent an
# abort from occurring? It seems wrong to swallow them
# all, yet you want to be sure that other abort logic is
# executed regardless.
try:
self._server.tpc_abort(id(txn))
self._call('tpc_abort', id(txn))
except ClientDisconnected:
logger.debug("%s ClientDisconnected in tpc_abort() ignored",
self.__name__)
finally:
self._tbuf.clear()
self._seriald.clear()
del self._serials[:]
self._iterator_gc()
self.end_transaction()
self.tpc_end(txn)
def tpc_finish(self, txn, f=None):
def tpc_finish(self, txn, f=lambda tid: None):
"""Storage API: finish a transaction."""
if txn is not self._transaction:
raise POSException.StorageTransactionError(
"tpc_finish called with wrong transaction")
self._load_lock.acquire()
try:
if self._midtxn_disconnect:
raise ClientDisconnected(
'Calling tpc_finish() on a disconnected transaction')
tbuf = self._check_trans(txn, 'tpc_finish')
finished = 0
try:
self._lock.acquire() # for atomic processing of invalidations
try:
tid = self._server.tpc_finish(id(txn))
finished = 1
self._update_cache(tid)
if f is not None:
f(tid)
finally:
self._lock.release()
r = self._check_serials()
assert r is None or len(r) == 0, "unhandled serialnos: %s" % r
except:
if finished:
# The server successfully committed. If we get a failure
# here, our own state will be in question, so reconnect.
self._connection.close()
raise
self.end_transaction()
tid = self._server.tpc_finish(id(txn), tbuf, f)
finally:
self._load_lock.release()
self.tpc_end(txn)
self._iterator_gc()
def _update_cache(self, tid):
"""Internal helper to handle objects modified by a transaction.
This iterates over the objects in the transaction buffer and
update or invalidate the cache.
self._update_blob_cache(tbuf, tid)
def _update_blob_cache(self, tbuf, tid):
"""Internal helper move blobs updated by a transaction to the cache.
"""
# Must be called with _lock already acquired.
# Not sure why _update_cache() would be called on a closed storage.
if self._cache is None:
return
for oid, _ in six.iteritems(self._seriald):
self._cache.invalidate(oid, tid)
for oid, data in self._tbuf:
# If data is None, we just invalidate.
if data is not None:
s = self._seriald[oid]
if s != ResolvedSerial:
assert s == tid, (s, tid)
self._cache.store(oid, s, None, data)
else:
# object deletion
self._cache.invalidate(oid, tid)
if self.fshelper is not None:
blobs = self._tbuf.blobs
blobs = tbuf.blobs
had_blobs = False
while blobs:
oid, blobfilename = blobs.pop()
......@@ -1263,9 +810,6 @@ class ClientStorage(object):
if had_blobs:
self._check_blob_size(self._blob_data_bytes_loaded)
self._cache.setLastTid(tid)
self._tbuf.clear()
def undo(self, trans_id, txn):
"""Storage API: undo a transaction.
......@@ -1276,12 +820,12 @@ class ClientStorage(object):
a storage.
"""
self._check_trans(txn)
self._server.undoa(trans_id, id(txn))
self._check_trans(txn, 'undo')
self._async('undoa', trans_id, id(txn))
def undoInfo(self, first=0, last=-20, specification=None):
"""Storage API: return undo information."""
return self._server.undoInfo(first, last, specification)
return self._call('undoInfo', first, last, specification)
def undoLog(self, first=0, last=-20, filter=None):
"""Storage API: return a sequence of TransactionDescription objects.
......@@ -1294,7 +838,7 @@ class ClientStorage(object):
"""
if filter is not None:
return []
return self._server.undoLog(first, last)
return self._call('undoLog', first, last)
# Recovery support
......@@ -1309,8 +853,8 @@ class ClientStorage(object):
def restore(self, oid, serial, data, version, prev_txn, transaction):
"""Write data already committed in a separate database."""
assert not version
self._check_trans(transaction)
self._server.restorea(oid, serial, data, prev_txn, id(transaction))
self._check_trans(transaction, 'restore')
self._async('restorea', oid, serial, data, prev_txn, id(transaction))
# Don't update the transaction buffer, because current data are
# unaffected.
return self._check_serials()
......@@ -1318,210 +862,27 @@ class ClientStorage(object):
# Below are methods invoked by the StorageServer
def serialnos(self, args):
"""Server callback to pass a list of changed (oid, serial) pairs."""
self._serials.extend(args)
"""Server callback to pass a list of changed (oid, serial) pairs.
"""
for oid, s in args:
self._tbuf.serial(oid, s)
def info(self, dict):
"""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")
if PY3:
self._pickler = Pickler(self._tfile, 3)
else:
self._pickler = Pickler(self._tfile, 1)
self._pickler.fast = 1 # Don't use the memo
if self._connection.peer_protocol_version < b'Z309':
client = ClientStorage308Adapter(self)
else:
client = self
# allow incoming invalidations:
self._connection.register_object(client)
# 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.
server_tid = server.lastTransaction()
if not self._cache:
logger.info("%s No verification necessary -- empty cache",
self.__name__)
if server_tid != utils.z64:
self._cache.setLastTid(server_tid)
self.finish_verification()
return "empty cache"
cache_tid = self._cache.getLastTid()
if cache_tid != utils.z64:
if server_tid == cache_tid:
logger.info(
"%s No verification necessary"
" (cache_tid up-to-date %r)",
self.__name__, server_tid)
self.finish_verification()
return "no verification"
elif server_tid < cache_tid:
message = ("%s Client has seen newer transactions than server!"
% self.__name__)
logger.critical(message)
raise ClientStorageError(message)
# log some hints about last transaction
logger.info("%s last inval tid: %r %s\n",
self.__name__, cache_tid,
tid2time(cache_tid))
logger.info("%s last transaction: %r %s",
self.__name__, server_tid,
server_tid and tid2time(server_tid))
pair = server.getInvalidations(cache_tid)
if pair is not None:
logger.info("%s Recovering %d invalidations",
self.__name__, len(pair[1]))
self.finish_verification(pair)
return "quick verification"
elif server_tid != utils.z64:
# Hm, to have gotten here, the cache is non-empty, but
# it has no last tid. This doesn't seem like good situation.
# We'll have to verify the cache, if we're willing.
self._cache.setLastTid(server_tid)
ZODB.event.notify(ZEO.interfaces.StaleCache(self))
# From this point on, we do not have complete information about
# the missed transactions. The reason is that cache
# verification only checks objects in the client cache and
# there may be objects in the object caches that aren't in the
# client cach that would need verification too. We avoid that
# problem by just invalidating the objects in the object caches.
if self._db is not None:
self._db.invalidateCache()
if self._cache and self._drop_cache_rather_verify:
logger.critical("%s dropping stale cache", self.__name__)
self._cache.clear()
if server_tid:
self._cache.setLastTid(server_tid)
self.finish_verification()
return "cache dropped"
logger.info("%s Verifying cache", self.__name__)
for oid, tid in self._cache.contents():
server.verify(oid, tid)
server.endZeoVerify()
return "full verification"
def invalidateVerify(self, oid):
"""Server callback to invalidate an oid pair.
This is called as part of cache validation.
"""
# Invalidation as result of verify_cache().
# Queue an invalidate for the end the verification procedure.
if self._pickler is None:
# This should never happen.
logger.error("%s invalidateVerify with no _pickler", self.__name__)
return
self._pickler.dump((None, [oid]))
def endVerify(self):
"""Server callback to signal end of cache validation."""
logger.info("%s endVerify finishing", self.__name__)
self.finish_verification()
logger.info("%s endVerify finished", self.__name__)
def finish_verification(self, catch_up=None):
self._lock.acquire()
try:
if catch_up:
# process catch-up invalidations
self._process_invalidations(*catch_up)
if self._pickler is None:
return
# write end-of-data marker
self._pickler.dump((None, None))
self._pickler = None
self._tfile.seek(0)
unpickler = Unpickler(self._tfile)
min_tid = self._cache.getLastTid()
while 1:
tid, oids = unpickler.load()
logger.debug('pickled inval %r %r', tid, min_tid)
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()
self._server = self._pending_server
self._ready.set()
self._pending_server = None
def invalidateTransaction(self, tid, oids):
"""Server callback: Invalidate objects modified by tid."""
self._lock.acquire()
try:
if self._pickler is not None:
logger.debug(
"%s Transactional invalidation during cache verification",
self.__name__)
self._pickler.dump((tid, oids))
else:
self._process_invalidations(tid, oids)
finally:
self._lock.release()
def _process_invalidations(self, tid, oids):
for oid in oids:
if oid == self._load_oid:
self._load_status = 0
self._cache.invalidate(oid, tid)
self._cache.setLastTid(tid)
if self._db is not None:
self._db.invalidate(tid, oids)
# The following are for compatibility with protocol version 2.0.0
def invalidateTrans(self, oids):
return self.invalidateTransaction(None, oids)
invalidate = invalidateVerify
end = endVerify
Invalidate = invalidateTrans
# IStorageIteration
def iterator(self, start=None, stop=None):
"""Return an IStorageTransactionInformation iterator."""
# iids are "iterator IDs" that can be used to query an iterator whose
# status is held on the server.
iid = self._server.iterator_start(start, stop)
iid = self._call('iterator_start', start, stop)
return self._setup_iterator(TransactionIterator, iid)
def _setup_iterator(self, factory, iid, *args):
......@@ -1558,10 +919,11 @@ class ClientStorage(object):
# objects waiting to be cleaned up. So there's never any risk
# of confusing TransactionIterator objects that are in use.
iids = self._iterator_ids - set(self._iterators)
self._iterators._last_gc = time.time() # let tests know we've been called
# let tests know we've been called:
self._iterators._last_gc = time.time()
if iids:
try:
self._server.iterator_gc(list(iids))
self._async('iterator_gc', list(iids))
except ClientDisconnected:
# If we get disconnected, all of the iterators on the
# server are thrown away. We should clear ours too:
......@@ -1569,8 +931,7 @@ class ClientStorage(object):
self._iterator_ids -= iids
def server_status(self):
return self._server.server_status()
return self._call('server_status')
class TransactionIterator(object):
......@@ -1589,7 +950,7 @@ class TransactionIterator(object):
if self._iid < 0:
raise ClientDisconnected("Disconnected iterator")
tx_data = self._storage._server.iterator_next(self._iid)
tx_data = self._storage._call('iterator_next', self._iid)
if tx_data is None:
# The iterator is exhausted, and the server has already
# disposed it.
......@@ -1619,8 +980,8 @@ class ClientStorageTransactionInformation(ZODB.BaseStorage.TransactionRecord):
self.extension = extension
def __iter__(self):
riid = self._storage._server.iterator_record_start(self._txiter._iid,
self.tid)
riid = self._storage._call('iterator_record_start',
self._txiter._iid, self.tid)
return self._storage._setup_iterator(RecordIterator, riid)
......@@ -1639,7 +1000,7 @@ class RecordIterator(object):
# We finished iteration once already and the server can't know
# about the iteration anymore.
raise StopIteration()
item = self._storage._server.iterator_record_next(self._riid)
item = self._storage._call('iterator_record_next', self._riid)
if item is None:
# The iterator is exhausted, and the server has already
# disposed it.
......@@ -1650,21 +1011,6 @@ class RecordIterator(object):
next = __next__
class ClientStorage308Adapter:
def __init__(self, client):
self.client = client
def invalidateTransaction(self, tid, args):
self.client.invalidateTransaction(tid, [arg[0] for arg in args])
def invalidateVerify(self, arg):
self.client.invalidateVerify(arg[0])
def __getattr__(self, name):
return getattr(self.client, name)
class BlobCacheLayout(object):
size = 997
......@@ -1807,3 +1153,16 @@ def _lock_blob(path):
raise
else:
break
def open_cache(cache, var, client, cache_size):
if isinstance(cache, (None.__class__, str)):
from ZEO.cache import ClientCache
if cache is None:
if client:
cache = os.path.join(var or os.getcwd(), client)
else:
return ClientCache(cache, cache_size)
cache = ClientCache(cache, cache_size)
return cache
src/ZEO/Exceptions.py
View file @ 3f31236b
......@@ -26,3 +26,7 @@ class ClientDisconnected(ClientStorageError):
class AuthError(StorageError):
"""The client provided invalid authentication credentials."""
class ProtocolError(ClientStorageError):
"""A client contacted a server with an incomparible protocol
"""
src/ZEO/ServerStub.py deleted 100644 → 0
View file @ bdbc36dd
##############################################################################
#
# Copyright (c) 2001, 2002, 2003 Zope Foundation and Contributors.
# All Rights Reserved.
#
# This software is subject to the provisions of the Zope Public License,
# Version 2.1 (ZPL). A copy of the ZPL should accompany this distribution.
# THIS SOFTWARE IS PROVIDED "AS IS" AND ANY AND ALL EXPRESS OR IMPLIED
# WARRANTIES ARE DISCLAIMED, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
# WARRANTIES OF TITLE, MERCHANTABILITY, AGAINST INFRINGEMENT, AND FITNESS
# FOR A PARTICULAR PURPOSE
#
##############################################################################
"""RPC stubs for interface exported by StorageServer."""
import time
from ZODB.utils import z64
##
# ZEO storage server.
# <p>
# Remote method calls can be synchronous or asynchronous. If the call
# is synchronous, the client thread blocks until the call returns. A
# single client can only have one synchronous request outstanding. If
# several threads share a single client, threads other than the caller
# will block only if the attempt to make another synchronous call.
# An asynchronous call does not cause the client thread to block. An
# exception raised by an asynchronous method is logged on the server,
# but is not returned to the client.
class StorageServer:
"""An RPC stub class for the interface exported by ClientStorage.
This is the interface presented by the StorageServer to the
ClientStorage; i.e. the ClientStorage calls these methods and they
are executed in the StorageServer.
See the StorageServer module for documentation on these methods,
with the exception of _update(), which is documented here.
"""
def __init__(self, rpc):
"""Constructor.
The argument is a connection: an instance of the
zrpc.connection.Connection class.
"""
self.rpc = rpc
def extensionMethod(self, name):
return ExtensionMethodWrapper(self.rpc, name).call
##
# Register current connection with a storage and a mode.
# In effect, it is like an open call.
# @param storage_name a string naming the storage. This argument
# is primarily for backwards compatibility with servers
# that supported multiple storages.
# @param read_only boolean
# @exception ValueError unknown storage_name or already registered
# @exception ReadOnlyError storage is read-only and a read-write
# connectio was requested
def register(self, storage_name, read_only):
self.rpc.call('register', storage_name, read_only)
##
# Return dictionary of meta-data about the storage.
# @defreturn dict
def get_info(self):
return self.rpc.call('get_info')
##
# Check whether the server requires authentication. Returns
# the name of the protocol.
# @defreturn string
def getAuthProtocol(self):
return self.rpc.call('getAuthProtocol')
##
# Return id of the last committed transaction
# @defreturn string
def lastTransaction(self):
# Not in protocol version 2.0.0; see __init__()
return self.rpc.call('lastTransaction') or z64
##
# Return invalidations for all transactions after tid.
# @param tid transaction id
# @defreturn 2-tuple, (tid, list)
# @return tuple containing the last committed transaction
# and a list of oids that were invalidated. Returns
# None and an empty list if the server does not have
# the list of oids available.
def getInvalidations(self, tid):
# Not in protocol version 2.0.0; see __init__()
return self.rpc.call('getInvalidations', tid)
##
# Check whether a serial number is current for oid.
# If the serial number is not current, the
# server will make an asynchronous invalidateVerify() call.
# @param oid object id
# @param s serial number
# @defreturn async
def zeoVerify(self, oid, s):
self.rpc.callAsync('zeoVerify', oid, s)
##
# Check whether current serial number is valid for oid.
# If the serial number is not current, the server will make an
# asynchronous invalidateVerify() call.
# @param oid object id
# @param serial client's current serial number
# @defreturn async
def verify(self, oid, serial):
self.rpc.callAsync('verify', oid, serial)
##
# Signal to the server that cache verification is done.
# @defreturn async
def endZeoVerify(self):
self.rpc.callAsync('endZeoVerify')
##
# Generate a new set of oids.
# @param n number of new oids to return
# @defreturn list
# @return list of oids
def new_oids(self, n=None):
if n is None:
return self.rpc.call('new_oids')
else:
return self.rpc.call('new_oids', n)
##
# Pack the storage.
# @param t pack time
# @param wait optional, boolean. If true, the call will not
# return until the pack is complete.
def pack(self, t, wait=None):
if wait is None:
self.rpc.call('pack', t)
else:
self.rpc.call('pack', t, wait)
##
# Return current data for oid.
# @param oid object id
# @defreturn 2-tuple
# @return 2-tuple, current non-version data, serial number
# @exception KeyError if oid is not found
def zeoLoad(self, oid):
return self.rpc.call('zeoLoad', oid)[:2]
##
# Return current data for oid, and the tid of the
# transaction that wrote the most recent revision.
# @param oid object id
# @defreturn 2-tuple
# @return data, transaction id
# @exception KeyError if oid is not found
def loadEx(self, oid):
return self.rpc.call("loadEx", oid)
##
# Return non-current data along with transaction ids that identify
# the lifetime of the specific revision.
# @param oid object id
# @param tid a transaction id that provides an upper bound on
# the lifetime of the revision. That is, loadBefore
# returns the revision that was current before tid committed.
# @defreturn 4-tuple
# @return data, serial numbr, start transaction id, end transaction id
def loadBefore(self, oid, tid):
return self.rpc.call("loadBefore", oid, tid)
##
# Storage new revision of oid.
# @param oid object id
# @param serial serial number that this transaction read
# @param data new data record for oid
# @param id id of current transaction
# @defreturn async
def storea(self, oid, serial, data, id):
self.rpc.callAsync('storea', oid, serial, data, id)
def checkCurrentSerialInTransaction(self, oid, serial, id):
self.rpc.callAsync('checkCurrentSerialInTransaction', oid, serial, id)
def restorea(self, oid, serial, data, prev_txn, id):
self.rpc.callAsync('restorea', oid, serial, data, prev_txn, id)
def storeBlob(self, oid, serial, data, blobfilename, txn):
# Store a blob to the server. We don't want to real all of
# the data into memory, so we use a message iterator. This
# allows us to read the blob data as needed.
def store():
yield ('storeBlobStart', ())
f = open(blobfilename, 'rb')
while 1:
chunk = f.read(59000)
if not chunk:
break
yield ('storeBlobChunk', (chunk, ))
f.close()
yield ('storeBlobEnd', (oid, serial, data, id(txn)))
self.rpc.callAsyncIterator(store())
def storeBlobShared(self, oid, serial, data, filename, id):
self.rpc.callAsync('storeBlobShared', oid, serial, data, filename, id)
def deleteObject(self, oid, serial, id):
self.rpc.callAsync('deleteObject', oid, serial, id)
##
# Start two-phase commit for a transaction
# @param id id used by client to identify current transaction. The
# only purpose of this argument is to distinguish among multiple
# threads using a single ClientStorage.
# @param user name of user committing transaction (can be "")
# @param description string containing transaction metadata (can be "")
# @param ext dictionary of extended metadata (?)
# @param tid optional explicit tid to pass to underlying storage
# @param status optional status character, e.g "p" for pack
# @defreturn async
def tpc_begin(self, id, user, descr, ext, tid, status):
self.rpc.callAsync('tpc_begin', id, user, descr, ext, tid, status)
def vote(self, trans_id):
return self.rpc.call('vote', trans_id)
def tpc_finish(self, id):
return self.rpc.call('tpc_finish', id)
def tpc_abort(self, id):
self.rpc.call('tpc_abort', id)
def history(self, oid, length=None):
if length is None:
return self.rpc.call('history', oid)
else:
return self.rpc.call('history', oid, length)
def record_iternext(self, next):
return self.rpc.call('record_iternext', next)
def sendBlob(self, oid, serial):
return self.rpc.call('sendBlob', oid, serial)
def getTid(self, oid):
return self.rpc.call('getTid', oid)
def loadSerial(self, oid, serial):
return self.rpc.call('loadSerial', oid, serial)
def new_oid(self):
return self.rpc.call('new_oid')
def undoa(self, trans_id, trans):
self.rpc.callAsync('undoa', trans_id, trans)
def undoLog(self, first, last):
return self.rpc.call('undoLog', first, last)
def undoInfo(self, first, last, spec):
return self.rpc.call('undoInfo', first, last, spec)
def iterator_start(self, start, stop):
return self.rpc.call('iterator_start', start, stop)
def iterator_next(self, iid):
return self.rpc.call('iterator_next', iid)
def iterator_record_start(self, txn_iid, tid):
return self.rpc.call('iterator_record_start', txn_iid, tid)
def iterator_record_next(self, iid):
return self.rpc.call('iterator_record_next', iid)
def iterator_gc(self, iids):
return self.rpc.callAsync('iterator_gc', iids)
def server_status(self):
return self.rpc.call("server_status")
def set_client_label(self, label):
return self.rpc.callAsync('set_client_label', label)
class StorageServer308(StorageServer):
def __init__(self, rpc):
if rpc.peer_protocol_version == b'Z200':
self.lastTransaction = lambda: z64
self.getInvalidations = lambda tid: None
self.getAuthProtocol = lambda: None
StorageServer.__init__(self, rpc)
def history(self, oid, length=None):
if length is None:
return self.rpc.call('history', oid, '')
else:
return self.rpc.call('history', oid, '', length)
def getInvalidations(self, tid):
# Not in protocol version 2.0.0; see __init__()
result = self.rpc.call('getInvalidations', tid)
if result is not None:
result = result[0], [oid for (oid, version) in result[1]]
return result
def verify(self, oid, serial):
self.rpc.callAsync('verify', oid, '', serial)
def loadEx(self, oid):
return self.rpc.call("loadEx", oid, '')[:2]
def storea(self, oid, serial, data, id):
self.rpc.callAsync('storea', oid, serial, data, '', id)
def storeBlob(self, oid, serial, data, blobfilename, txn):
# Store a blob to the server. We don't want to real all of
# the data into memory, so we use a message iterator. This
# allows us to read the blob data as needed.
def store():
yield ('storeBlobStart', ())
f = open(blobfilename, 'rb')
while 1:
chunk = f.read(59000)
if not chunk:
break
yield ('storeBlobChunk', (chunk, ))
f.close()
yield ('storeBlobEnd', (oid, serial, data, '', id(txn)))
self.rpc.callAsyncIterator(store())
def storeBlobShared(self, oid, serial, data, filename, id):
self.rpc.callAsync('storeBlobShared', oid, serial, data, filename,
'', id)
def zeoVerify(self, oid, s):
self.rpc.callAsync('zeoVerify', oid, s, None)
def iterator_start(self, start, stop):
raise NotImplementedError
def iterator_next(self, iid):
raise NotImplementedError
def iterator_record_start(self, txn_iid, tid):
raise NotImplementedError
def iterator_record_next(self, iid):
raise NotImplementedError
def iterator_gc(self, iids):
raise NotImplementedError
def stub(client, connection):
start = time.time()
# Wait until we know what version the other side is using.
while connection.peer_protocol_version is None:
if time.time()-start > 10:
raise ValueError("Timeout waiting for protocol handshake")
time.sleep(0.1)
if connection.peer_protocol_version < b'Z309':
return StorageServer308(connection)
return StorageServer(connection)
class ExtensionMethodWrapper:
def __init__(self, rpc, name):
self.rpc = rpc
self.name = name
def call(self, *a, **kwa):
return self.rpc.call(self.name, *a, **kwa)
src/ZEO/StorageServer.py
View file @ 3f31236b
......@@ -1574,10 +1574,6 @@ class ZEOStorage308Adapter:
abortVersion = commitVersion
def zeoLoad(self, oid): # Z200
p, s = self.storage.loadEx(oid)
return p, s, '', None, None
def __getattr__(self, name):
return getattr(self.storage, name)
......
src/ZEO/TransactionBuffer.py
View file @ 3f31236b
......@@ -21,44 +21,24 @@ is used to store the data until a commit or abort.
# A faster implementation might store trans data in memory until it
# reaches a certain size.
from threading import Lock
import os
import tempfile
import ZODB.blob
from ZODB.ConflictResolution import ResolvedSerial
from ZEO._compat import Pickler, Unpickler
class TransactionBuffer:
# Valid call sequences:
#
# ((store | invalidate)* begin_iterate next* clear)* close
#
# get_size can be called any time
# The TransactionBuffer is used by client storage to hold update
# data until the tpc_finish(). It is normally used by a single
# data until the tpc_finish(). It is only used by a single
# thread, because only one thread can be in the two-phase commit
# at one time.
# It is possible, however, for one thread to close the storage
# while another thread is in the two-phase commit. We must use
# a lock to guard against this race, because unpredictable things
# can happen in Python if one thread closes a file that another
# thread is reading. In a debug build, an assert() can fail.
# Caution: If an operation is performed on a closed TransactionBuffer,
# it has no effect and does not raise an exception. The only time
# this should occur is when a ClientStorage is closed in one
# thread while another thread is in its tpc_finish(). It's not
# clear what should happen in this case. If the tpc_finish()
# completes without error, the Connection using it could have
# inconsistent data. This should have minimal effect, though,
# because the Connection is connected to a closed storage.
def __init__(self):
def __init__(self, connection_generation):
self.connection_generation = connection_generation
self.file = tempfile.TemporaryFile(suffix=".tbuf")
self.lock = Lock()
self.closed = 0
self.count = 0
self.size = 0
self.blobs = []
......@@ -66,89 +46,45 @@ class TransactionBuffer:
# stored are builtin types -- strings or None.
self.pickler = Pickler(self.file, 1)
self.pickler.fast = 1
self.serials = {} # processed { oid -> serial }
self.exception = None
def close(self):
self.clear()
self.lock.acquire()
try:
self.closed = 1
try:
self.file.close()
except OSError:
pass
finally:
self.lock.release()
def store(self, oid, data):
"""Store oid, version, data for later retrieval"""
self.lock.acquire()
try:
if self.closed:
return
self.pickler.dump((oid, data))
self.count += 1
# Estimate per-record cache size
self.size = self.size + (data and len(data) or 0) + 31
finally:
self.lock.release()
def serial(self, oid, serial):
if isinstance(serial, Exception):
self.exception = serial
else:
self.serials[oid] = serial
def storeBlob(self, oid, blobfilename):
self.blobs.append((oid, blobfilename))
def invalidate(self, oid):
self.lock.acquire()
try:
if self.closed:
return
self.pickler.dump((oid, None))
self.count += 1
finally:
self.lock.release()
def clear(self):
"""Mark the buffer as empty"""
self.lock.acquire()
try:
if self.closed:
return
self.file.seek(0)
self.count = 0
self.size = 0
while self.blobs:
oid, blobfilename = self.blobs.pop()
if os.path.exists(blobfilename):
ZODB.blob.remove_committed(blobfilename)
finally:
self.lock.release()
def __iter__(self):
self.lock.acquire()
try:
if self.closed:
return
self.file.flush()
self.file.seek(0)
return TBIterator(self.file, self.count)
finally:
self.lock.release()
class TBIterator(object):
def __init__(self, f, count):
self.file = f
self.count = count
self.unpickler = Unpickler(f)
def __iter__(self):
return self
def __next__(self):
"""Return next tuple of data or None if EOF"""
if self.count == 0:
self.file.seek(0)
self.size = 0
raise StopIteration
oid_ver_data = self.unpickler.load()
self.count -= 1
return oid_ver_data
next = __next__
unpickler = Unpickler(self.file)
serials = self.serials
# Gaaaa, this is awkward. There can be entries in serials that
# aren't in the buffer, because undo. Entries can be repeated
# in the buffer, because ZODB. (Maybe this is a bug now, but
# it may be a feature later.
seen = set()
for i in range(self.count):
oid, data = unpickler.load()
seen.add(oid)
yield oid, data, serials[oid] == ResolvedSerial
# We may have leftover serials because undo
for oid, serial in serials.items():
if oid not in seen:
yield oid, None, serial == ResolvedSerial
src/ZEO/asyncio/client.py
View file @ 3f31236b
......@@ -6,6 +6,7 @@ import concurrent.futures
import logging
import random
import threading
import traceback
import ZEO.Exceptions
import ZODB.POSException
......@@ -73,7 +74,8 @@ class Protocol(asyncio.Protocol):
if self.transport is not None:
self.transport.close()
for future in self.futures.values():
future.set_exception(Closed())
future.set_exception(
ZEO.Exceptions.ClientDisconnected("Closed"))
self.futures.clear()
def protocol_factory(self):
......@@ -156,8 +158,7 @@ class Protocol(asyncio.Protocol):
return self.transport.get_extra_info('peername')
def connection_lost(self, exc):
if exc is None:
# we were closed
if self.closed:
for f in self.futures.values():
f.cancel()
else:
......@@ -320,7 +321,8 @@ class Client:
# connect.
protocol = None
ready = False
ready = None # Tri-value: None=Never connected, True=connected,
# False=Disconnected
def __init__(self, loop,
addrs, client, cache, storage_key, read_only, connect_poll,
......@@ -350,6 +352,8 @@ class Client:
def close(self):
if not self.closed:
self.closed = True
self.ready = False
if self.protocol is not None:
self.protocol.close()
self.cache.close()
self._clear_protocols()
......@@ -364,6 +368,7 @@ class Client:
if protocol is None or protocol is self.protocol:
if protocol is self.protocol and protocol is not None:
self.client.notify_disconnected()
if self.ready:
self.ready = False
self.connected = concurrent.futures.Future()
self.protocol = None
......@@ -468,8 +473,8 @@ class Client:
@self.protocol.promise('get_info')
def got_info(info):
self.connected.set_result(None)
self.client.notify_connected(self, info)
self.connected.set_result(None)
@got_info.catch
def failed_info(exc):
......@@ -497,11 +502,16 @@ class Client:
def _when_ready(self, func, result_future, *args):
if self.ready is None:
# We started without waiting for a connection. (prob tests :( )
result_future.set_exception(
ZEO.Exceptions.ClientDisconnected("never connected"))
else:
@self.connected.add_done_callback
def done(future):
e = future.exception()
if e is not None:
result_future.set_exception(e)
future.set_exception(e)
else:
if self.ready:
func(result_future, *args)
......@@ -558,7 +568,7 @@ class Client:
cache.store(oid, tid, None, data)
cache.setLastTid(tid)
f(tid)
future.set_result(None)
future.set_result(tid)
committed.catch(future.set_exception)
else:
......@@ -579,6 +589,17 @@ class Client:
def protocol_version(self):
return self.protocol.protocol_version
def is_read_only(self):
try:
protocol = self.protocol
except AttributeError:
return self.read_only
else:
if protocol is None:
return self.read_only
else:
return protocol.read_only
class ClientRunner:
def set_options(self, addrs, wrapper, cache, storage_key, read_only,
......@@ -591,6 +612,8 @@ class ClientRunner:
def setup_delegation(self, loop):
self.loop = loop
self.client = Client(loop, *self.__args)
self.call_threadsafe = self.client.call_threadsafe
self.call_async_threadsafe = self.client.call_async_threadsafe
from concurrent.futures import Future
call_soon_threadsafe = loop.call_soon_threadsafe
......@@ -614,10 +637,17 @@ class ClientRunner:
return future.result(self.timeout if timeout is False else timeout)
def call(self, method, *args, timeout=None):
return self.__call(self.client.call_threadsafe, method, args)
return self.__call(self.call_threadsafe, method, args)
def call_future(self, method, *args):
# for tests
result = concurrent.futures.Future()
self.loop.call_soon_threadsafe(
self.call_threadsafe, result, method, args)
return result
def async(self, method, *args):
return self.__call(self.client.call_async_threadsafe, method, args)
return self.__call(self.call_async_threadsafe, method, args)
def async_iter(self, it):
return self.__call(self.client.call_async_iter_threadsafe, it)
......@@ -648,6 +678,12 @@ class ClientRunner:
def close(self):
self.__call(self.client.close_threadsafe)
# Short circuit from now on. We're closed.
def call_closed(*a, **k):
raise ZEO.Exceptions.ClientDisconnected('closed')
self.__call = call_closed
def new_addr(self, addrs):
# This usually doesn't have an immediate effect, since the
# addrs aren't used until the client disconnects.xs
......@@ -663,21 +699,45 @@ class ClientThread(ClientRunner):
def __init__(self, addrs, client, cache,
storage_key='1', read_only=False, timeout=30,
disconnect_poll=1):
disconnect_poll=1, wait=True):
self.set_options(addrs, client, cache, storage_key, read_only,
timeout, disconnect_poll)
threading.Thread(
self.thread = threading.Thread(
target=self.run,
name='zeo_client_'+storage_key,
daemon=True,
).start()
)
self.started = threading.Event()
self.thread.start()
self.started.wait()
if wait:
self.connected.result(timeout)
exception = None
def run(self):
try:
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
self.setup_delegation(loop)
self.started.set()
loop.run_forever()
except Exception as exc:
logger.exception("Client thread")
self.exception = exc
raise
else:
loop.close()
logger.debug('Stopping client thread')
closed = False
def close(self):
if not self.closed:
self.closed = True
super().close()
self.loop.call_soon_threadsafe(self.loop.stop)
self.thread.join(9)
if self.exception:
raise self.exception
class Promise:
"""Lightweight future with a partial promise API.
......
src/ZEO/asyncio/testing.py
View file @ 3f31236b
......@@ -102,3 +102,12 @@ class Transport:
def get_extra_info(self, name):
return self.extra[name]
class AsyncRPC:
"""Adapt an asyncio API to an RPC to help hysterical tests
"""
def __init__(self, api):
self.api = api
def __getattr__(self, name):
return lambda *a, **kw: self.api.call(name, *a, **kw)
src/ZEO/tests/CommitLockTests.py
View file @ 3f31236b
......@@ -60,17 +60,11 @@ class WorkerThread(TestThread):
# coordinate the action of multiple threads that all call
# vote(). This method sends the vote call, then sets the
# event saying vote was called, then waits for the vote
# response. It digs deep into the implementation of the client.
# response.
# This method is a replacement for:
# self.ready.set()
# self.storage.tpc_vote(self.trans)
rpc = self.storage._server.rpc
msgid = rpc._deferred_call('vote', id(self.trans))
future = self.storage._server.call_future('vote', id(self.trans))
self.ready.set()
rpc._deferred_wait(msgid)
self.storage._check_serials()
future.result(9)
class CommitLockTests:
......
src/ZEO/tests/ConnectionTests.py
View file @ 3f31236b
......@@ -19,7 +19,6 @@ import asyncore
import threading
import logging
import ZEO.ServerStub
from ZEO.ClientStorage import ClientStorage
from ZEO.Exceptions import ClientDisconnected
from ZEO.zrpc.marshal import encode
......@@ -40,20 +39,10 @@ logger = logging.getLogger('ZEO.tests.ConnectionTests')
ZERO = '\0'*8
class TestServerStub(ZEO.ServerStub.StorageServer):
__super_getInvalidations = ZEO.ServerStub.StorageServer.getInvalidations
def getInvalidations(self, tid):
# squirrel the results away for inspection by test case
self._last_invals = self.__super_getInvalidations(tid)
return self._last_invals
class TestClientStorage(ClientStorage):
test_connection = False
StorageServerStubClass = TestServerStub
connection_count_for_tests = 0
def notifyConnected(self, conn):
......@@ -592,7 +581,6 @@ class InvqTests(CommonSetupTearDown):
def checkQuickVerificationWith2Clients(self):
perstorage = self.openClientStorage(cache="test", cache_size=4000)
self.assertEqual(perstorage.verify_result, "empty cache")
self._storage = self.openClientStorage()
oid = self._storage.new_oid()
......@@ -624,8 +612,6 @@ class InvqTests(CommonSetupTearDown):
label="perstorage.verify_result to be quick verification")
self.assertEqual(perstorage.verify_result, "quick verification")
self.assertEqual(perstorage._server._last_invals,
(revid, [oid]))
self.assertEqual(perstorage.load(oid, ''),
self._storage.load(oid, ''))
......
src/ZEO/tests/IterationTests.py
View file @ 3f31236b
......@@ -17,6 +17,8 @@ import transaction
import six
import gc
from ..asyncio.testing import AsyncRPC
class IterationTests:
def _assertIteratorIdsEmpty(self):
......@@ -52,7 +54,7 @@ class IterationTests:
def checkIteratorGCProtocol(self):
# Test garbage collection on protocol level.
server = self._storage._server
server = AsyncRPC(self._storage._server)
iid = server.iterator_start(None, None)
# None signals the end of iteration.
......@@ -79,7 +81,7 @@ class IterationTests:
self.assertEquals(0, len(self._storage._iterator_ids))
# The iterator has run through, so the server has already disposed it.
self.assertRaises(KeyError, self._storage._server.iterator_next, iid)
self.assertRaises(KeyError, self._storage._call, 'iterator_next', iid)
def checkIteratorGCSpanTransactions(self):
# Keep a hard reference to the iterator so it won't be automatically
......@@ -112,7 +114,7 @@ class IterationTests:
self._storage._iterators._last_gc = -1
self._dostore()
self._assertIteratorIdsEmpty()
self.assertRaises(KeyError, self._storage._server.iterator_next, iid)
self.assertRaises(KeyError, self._storage._call, 'iterator_next', iid)
def checkIteratorGCStorageTPCAborting(self):
# The odd little jig we do below arises from the fact that the
......@@ -129,7 +131,7 @@ class IterationTests:
self._storage.tpc_begin(t)
self._storage.tpc_abort(t)
self._assertIteratorIdsEmpty()
self.assertRaises(KeyError, self._storage._server.iterator_next, iid)
self.assertRaises(KeyError, self._storage._call, 'iterator_next', iid)
def checkIteratorGCStorageDisconnect(self):
......@@ -146,7 +148,7 @@ class IterationTests:
# Show that after disconnecting, the client side GCs the iterators
# as well. I'm calling this directly to avoid accidentally
# calling tpc_abort implicitly.
self._storage.notifyDisconnected()
self._storage.notify_disconnected()
self.assertEquals(0, len(self._storage._iterator_ids))
def checkIteratorParallel(self):
......
src/ZEO/tests/ThreadTests.py
View file @ 3f31236b
......@@ -17,7 +17,7 @@ import threading
import transaction
from ZODB.tests.StorageTestBase import zodb_pickle, MinPO
import ZEO.ClientStorage
import ZEO.Exceptions
ZERO = '\0'*8
......@@ -54,7 +54,7 @@ class GetsThroughVoteThread(BasicThread):
self.doNextEvent.wait(10)
try:
self.storage.tpc_finish(self.trans)
except ZEO.ClientStorage.ClientStorageError:
except ZEO.Exceptions.ClientStorageError:
self.gotValueError = 1
self.storage.tpc_abort(self.trans)
......@@ -67,7 +67,7 @@ class GetsThroughBeginThread(BasicThread):
def run(self):
try:
self.storage.tpc_begin(self.trans)
except ZEO.ClientStorage.ClientStorageError:
except ZEO.Exceptions.ClientStorageError:
self.gotValueError = 1
......
src/ZEO/tests/auth_plaintext.py deleted 100644 → 0
View file @ bdbc36dd
##############################################################################
#
# Copyright (c) 2003 Zope Foundation and Contributors.
# All Rights Reserved.
#
# This software is subject to the provisions of the Zope Public License,
# Version 2.1 (ZPL). A copy of the ZPL should accompany this distribution.
# THIS SOFTWARE IS PROVIDED "AS IS" AND ANY AND ALL EXPRESS OR IMPLIED
# WARRANTIES ARE DISCLAIMED, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
# WARRANTIES OF TITLE, MERCHANTABILITY, AGAINST INFRINGEMENT, AND FITNESS
# FOR A PARTICULAR PURPOSE
#
##############################################################################
"""Implements plaintext password authentication. The password is stored in
an SHA hash in the Database. The client sends over the plaintext
password, and the SHA hashing is done on the server side.
This mechanism offers *no network security at all*; the only security
is provided by not storing plaintext passwords on disk.
"""
from ZEO.hash import sha1
from ZEO.StorageServer import ZEOStorage
from ZEO.auth import register_module
from ZEO.auth.base import Client, Database
def session_key(username, realm, password):
key = "%s:%s:%s" % (username, realm, password)
return sha1(key.encode('utf-8')).hexdigest().encode('ascii')
class StorageClass(ZEOStorage):
def auth(self, username, password):
try:
dbpw = self.database.get_password(username)
except LookupError:
return 0
password_dig = sha1(password.encode('utf-8')).hexdigest()
if dbpw == password_dig:
self.connection.setSessionKey(session_key(username,
self.database.realm,
password))
return self._finish_auth(dbpw == password_dig)
class PlaintextClient(Client):
extensions = ["auth"]
def start(self, username, realm, password):
if self.stub.auth(username, password):
return session_key(username, realm, password)
else:
return None
register_module("plaintext", StorageClass, PlaintextClient, Database)
src/ZEO/tests/testAuth.py deleted 100644 → 0
View file @ bdbc36dd
##############################################################################
#
# Copyright (c) 2003 Zope Foundation and Contributors.
# All Rights Reserved.
#
# This software is subject to the provisions of the Zope Public License,
# Version 2.1 (ZPL). A copy of the ZPL should accompany this distribution.
# THIS SOFTWARE IS PROVIDED "AS IS" AND ANY AND ALL EXPRESS OR IMPLIED
# WARRANTIES ARE DISCLAIMED, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
# WARRANTIES OF TITLE, MERCHANTABILITY, AGAINST INFRINGEMENT, AND FITNESS
# FOR A PARTICULAR PURPOSE
#
##############################################################################
"""Test suite for AuthZEO."""
import os
import tempfile
import time
import unittest
from ZEO import zeopasswd
from ZEO.Exceptions import ClientDisconnected
from ZEO.tests.ConnectionTests import CommonSetupTearDown
class _AuthTest(CommonSetupTearDown):
__super_getServerConfig = CommonSetupTearDown.getServerConfig
__super_setUp = CommonSetupTearDown.setUp
__super_tearDown = CommonSetupTearDown.tearDown
realm = None
def setUp(self):
fd, self.pwfile = tempfile.mkstemp('pwfile')
os.close(fd)
if self.realm:
self.pwdb = self.dbclass(self.pwfile, self.realm)
else:
self.pwdb = self.dbclass(self.pwfile)
self.pwdb.add_user("foo", "bar")
self.pwdb.save()
self._checkZEOpasswd()
self.__super_setUp()
def _checkZEOpasswd(self):
args = ["-f", self.pwfile, "-p", self.protocol]
if self.protocol == "plaintext":
from ZEO.auth.base import Database
zeopasswd.main(args + ["-d", "foo"], Database)
zeopasswd.main(args + ["foo", "bar"], Database)
else:
zeopasswd.main(args + ["-d", "foo"])
zeopasswd.main(args + ["foo", "bar"])
def tearDown(self):
os.remove(self.pwfile)
self.__super_tearDown()
def getConfig(self, path, create, read_only):
return "<mappingstorage 1/>"
def getServerConfig(self, addr, ro_svr):
zconf = self.__super_getServerConfig(addr, ro_svr)
zconf.authentication_protocol = self.protocol
zconf.authentication_database = self.pwfile
zconf.authentication_realm = self.realm
return zconf
def wait(self):
for i in range(25):
time.sleep(0.1)
if self._storage.test_connection:
return
self.fail("Timed out waiting for client to authenticate")
def testOK(self):
# Sleep for 0.2 seconds to give the server some time to start up
# seems to be needed before and after creating the storage
self._storage = self.openClientStorage(wait=0, username="foo",
password="bar", realm=self.realm)
self.wait()
self.assert_(self._storage._connection)
self._storage._connection.poll()
self.assert_(self._storage.is_connected())
# Make a call to make sure the mechanism is working
self._storage.undoInfo()
def testNOK(self):
self._storage = self.openClientStorage(wait=0, username="foo",
password="noogie",
realm=self.realm)
self.wait()
# If the test established a connection, then it failed.
self.failIf(self._storage._connection)
def testUnauthenticatedMessage(self):
# Test that an unauthenticated message is rejected by the server
# if it was sent after the connection was authenticated.
self._storage = self.openClientStorage(wait=0, username="foo",
password="bar", realm=self.realm)
# Sleep for 0.2 seconds to give the server some time to start up
# seems to be needed before and after creating the storage
self.wait()
self._storage.undoInfo()
# Manually clear the state of the hmac connection
self._storage._connection._SizedMessageAsyncConnection__hmac_send = None
# Once the client stops using the hmac, it should be disconnected.
self.assertRaises(ClientDisconnected, self._storage.undoInfo)
class PlainTextAuth(_AuthTest):
import ZEO.tests.auth_plaintext
protocol = "plaintext"
database = "authdb.sha"
dbclass = ZEO.tests.auth_plaintext.Database
realm = "Plaintext Realm"
class DigestAuth(_AuthTest):
import ZEO.auth.auth_digest
protocol = "digest"
database = "authdb.digest"
dbclass = ZEO.auth.auth_digest.DigestDatabase
realm = "Digest Realm"
test_classes = [PlainTextAuth, DigestAuth]
def test_suite():
suite = unittest.TestSuite()
for klass in test_classes:
sub = unittest.makeSuite(klass)
suite.addTest(sub)
return suite
if __name__ == "__main__":
unittest.main(defaultTest='test_suite')
src/ZEO/tests/testZEO.py
View file @ 3f31236b
......@@ -46,7 +46,6 @@ import threading
import time
import transaction
import unittest
import ZEO.ServerStub
import ZEO.StorageServer
import ZEO.tests.ConnectionTests
import ZEO.zrpc.connection
......@@ -1721,7 +1720,7 @@ def can_use_empty_string_for_local_host_on_client():
"""
slow_test_classes = [
BlobAdaptedFileStorageTests, BlobWritableCacheTests,
#BlobAdaptedFileStorageTests, BlobWritableCacheTests,
MappingStorageTests, DemoStorageTests,
FileStorageTests, FileStorageHexTests, FileStorageClientHexTests,
]
......@@ -1733,12 +1732,6 @@ quick_test_classes = [
class ServerManagingClientStorage(ClientStorage):
class StorageServerStubClass(ZEO.ServerStub.StorageServer):
# Wait for abort for the benefit of blob_transaction.txt
def tpc_abort(self, id):
self.rpc.call('tpc_abort', id)
def __init__(self, name, blob_dir, shared=False, extrafsoptions=''):
if shared:
server_blob_dir = blob_dir
......
src/ZEO/tests/zeoserver.py
View file @ 3f31236b
......@@ -174,9 +174,6 @@ def main():
zo.realize(["-C", configfile])
addr = zo.address
if zo.auth_protocol == "plaintext":
__import__('ZEO.tests.auth_plaintext')
if isinstance(addr, tuple):
test_addr = addr[0], addr[1]+1
else:
......
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