StorageServer.py

##############################################################################
#
# Copyright (c) 2001, 2002, 2003 Zope Corporation 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
#
##############################################################################
"""The StorageServer class and the exception that it may raise.

This server acts as a front-end for one or more real storages, like
file storage or Berkeley storage.

TODO:  Need some basic access control-- a declaration of the methods
exported for invocation by the server.
"""

import asyncore
import cPickle
import os
import sys
import threading
import time
import logging

import transaction

from ZEO import ClientStub
from ZEO.CommitLog import CommitLog
from ZEO.monitor import StorageStats, StatsServer
from ZEO.zrpc.server import Dispatcher
from ZEO.zrpc.connection import ManagedServerConnection, Delay, MTDelay
from ZEO.zrpc.trigger import trigger
from ZEO.Exceptions import AuthError

from ZODB.ConflictResolution import ResolvedSerial
from ZODB.POSException import StorageError, StorageTransactionError
from ZODB.POSException import TransactionError, ReadOnlyError, ConflictError
from ZODB.serialize import referencesf
from ZODB.utils import u64, oid_repr
from ZODB.loglevels import BLATHER

logger = logging.getLogger('ZEO.StorageServer')

# TODO:  This used to say "ZSS", which is now implied in the logger name.
# Can this be either set to str(os.getpid()) (if that makes sense) or removed?
_label = "" # default label used for logging.

def set_label():
    """Internal helper to reset the logging label (e.g. after fork())."""
    global _label
    _label = "%s" % os.getpid()

def log(message, level=logging.INFO, label=None, exc_info=False):
    """Internal helper to log a message."""
    label = label or _label
    if label:
        message = "(%s) %s" % (label, message)
    logger.log(level, message, exc_info=exc_info)

class StorageServerError(StorageError):
    """Error reported when an unpicklable exception is raised."""

class ZEOStorage:
    """Proxy to underlying storage for a single remote client."""

    # Classes we instantiate.  A subclass might override.
    ClientStorageStubClass = ClientStub.ClientStorage

    # A list of extension methods.  A subclass with extra methods
    # should override.
    extensions = []

    def __init__(self, server, read_only=0, auth_realm=None):
        self.server = server
        # timeout and stats will be initialized in register()
        self.timeout = None
        self.stats = None
        self.connection = None
        self.client = None
        self.storage = None
        self.storage_id = "uninitialized"
        self.transaction = None
        self.read_only = read_only
        self.locked = 0
        self.verifying = 0
        self.store_failed = 0
        self.log_label = _label
        self.authenticated = 0
        self.auth_realm = auth_realm
        # The authentication protocol may define extra methods.
        self._extensions = {}
        for func in self.extensions:
            self._extensions[func.func_name] = None

    def finish_auth(self, authenticated):
        if not self.auth_realm:
            return 1
        self.authenticated = authenticated
        return authenticated

    def set_database(self, database):
        self.database = database

    def notifyConnected(self, conn):
        self.connection = conn # For restart_other() below
        self.client = self.ClientStorageStubClass(conn)
        addr = conn.addr
        if isinstance(addr, type("")):
            label = addr
        else:
            host, port = addr
            label = str(host) + ":" + str(port)
        self.log_label = _label + "/" + label

    def notifyDisconnected(self):
        # When this storage closes, we must ensure that it aborts
        # any pending transaction.
        if self.transaction is not None:
            self.log("disconnected during transaction %s" % self.transaction)
            self._abort()
        else:
            self.log("disconnected")
        if self.stats is not None:
            self.stats.clients -= 1

    def __repr__(self):
        tid = self.transaction and repr(self.transaction.id)
        if self.storage:
            stid = (self.storage._transaction and
                    repr(self.storage._transaction.id))
        else:
            stid = None
        name = self.__class__.__name__
        return "<%s %X trans=%s s_trans=%s>" % (name, id(self), tid, stid)

    def log(self, msg, level=logging.INFO, exc_info=False):
        log(msg, level=level, label=self.log_label, exc_info=exc_info)

    def setup_delegation(self):
        """Delegate several methods to the storage"""
        self.versionEmpty = self.storage.versionEmpty
        self.versions = self.storage.versions
        self.getSerial = self.storage.getSerial
        self.history = self.storage.history
        self.load = self.storage.load
        self.loadSerial = self.storage.loadSerial
        self.modifiedInVersion = self.storage.modifiedInVersion
        record_iternext = getattr(self.storage, 'record_iternext', None)
        if record_iternext is not None:
            self.record_iternext = record_iternext
            
            
        try:
            fn = self.storage.getExtensionMethods
        except AttributeError:
            # We must be running with a ZODB which
            # predates adding getExtensionMethods to
            # BaseStorage. Eventually this try/except
            # can be removed
            pass
        else:
            d = fn()
            self._extensions.update(d)
            for name in d.keys():
                assert not hasattr(self, name)
                setattr(self, name, getattr(self.storage, name))
        self.lastTransaction = self.storage.lastTransaction

    def _check_tid(self, tid, exc=None):
        if self.read_only:
            raise ReadOnlyError()
        if self.transaction is None:
            caller = sys._getframe().f_back.f_code.co_name
            self.log("no current transaction: %s()" % caller,
                     level=logging.WARNING)
            if exc is not None:
                raise exc(None, tid)
            else:
                return 0
        if self.transaction.id != tid:
            caller = sys._getframe().f_back.f_code.co_name
            self.log("%s(%s) invalid; current transaction = %s" %
                     (caller, repr(tid), repr(self.transaction.id)),
                     logging.WARNING)
            if exc is not None:
                raise exc(self.transaction.id, tid)
            else:
                return 0
        return 1

    def getAuthProtocol(self):
        """Return string specifying name of authentication module to use.

        The module name should be auth_%s where %s is auth_protocol."""
        protocol = self.server.auth_protocol
        if not protocol or protocol == 'none':
            return None
        return protocol

    def register(self, storage_id, read_only):
        """Select the storage that this client will use

        This method must be the first one called by the client.
        For authenticated storages this method will be called by the client
        immediately after authentication is finished.
        """
        if self.auth_realm and not self.authenticated:
            raise AuthError("Client was never authenticated with server!")

        if self.storage is not None:
            self.log("duplicate register() call")
            raise ValueError("duplicate register() call")
        storage = self.server.storages.get(storage_id)
        if storage is None:
            self.log("unknown storage_id: %s" % storage_id)
            raise ValueError("unknown storage: %s" % storage_id)

        if not read_only and (self.read_only or storage.isReadOnly()):
            raise ReadOnlyError()

        self.read_only = self.read_only or read_only
        self.storage_id = storage_id
        self.storage = storage
        self.setup_delegation()
        self.timeout, self.stats = self.server.register_connection(storage_id,
                                                                   self)

    def get_info(self):
        return {'length': len(self.storage),
                'size': self.storage.getSize(),
                'name': self.storage.getName(),
                'supportsUndo': self.storage.supportsUndo(),
                'supportsVersions': self.storage.supportsVersions(),
                'extensionMethods': self.getExtensionMethods(),
                'supports_record_iternext': hasattr(self, 'record_iternext'),
                }

    def get_size_info(self):
        return {'length': len(self.storage),
                'size': self.storage.getSize(),
                }

    def getExtensionMethods(self):
        return self._extensions

    def loadEx(self, oid, version):
        self.stats.loads += 1
        return self.storage.loadEx(oid, version)

    def loadBefore(self, oid, tid):
        self.stats.loads += 1
        return self.storage.loadBefore(oid, tid)

    def zeoLoad(self, oid):
        self.stats.loads += 1
        v = self.storage.modifiedInVersion(oid)
        if v:
            pv, sv = self.storage.load(oid, v)
        else:
            pv = sv = None
        try:
            p, s = self.storage.load(oid, '')
        except KeyError:
            if sv:
                # Created in version, no non-version data
                p = s = None
            else:
                raise
        return p, s, v, pv, sv

    def getInvalidations(self, tid):
        invtid, invlist = self.server.get_invalidations(tid)
        if invtid is None:
            return None
        self.log("Return %d invalidations up to tid %s"
                 % (len(invlist), u64(invtid)))
        return invtid, invlist

    def verify(self, oid, version, tid):
        try:
            t = self.storage.getTid(oid)
        except KeyError:
            self.client.invalidateVerify((oid, ""))
        else:
            if tid != t:
                # This will invalidate non-version data when the
                # client only has invalid version data.  Since this is
                # an uncommon case, we avoid the cost of checking
                # whether the serial number matches the current
                # non-version data.
                self.client.invalidateVerify((oid, version))

    def zeoVerify(self, oid, s, sv):
        if not self.verifying:
            self.verifying = 1
            self.stats.verifying_clients += 1
        try:
            os = self.storage.getTid(oid)
        except KeyError:
            self.client.invalidateVerify((oid, ''))
            # It's not clear what we should do now.  The KeyError
            # could be caused by an object uncreation, in which case
            # invalidation is right.  It could be an application bug
            # that left a dangling reference, in which case it's bad.
        else:
            # If the client has version data, the logic is a bit more
            # complicated.  If the current serial number matches the
            # client serial number, then the non-version data must
            # also be valid.  If the current serialno is for a
            # version, then the non-version data can't change.

            # If the version serialno isn't valid, then the
            # non-version serialno may or may not be valid.  Rather
            # than trying to figure it whether it is valid, we just
            # invalidate it.  Sending an invalidation for the
            # non-version data implies invalidating the version data
            # too, since an update to non-version data can only occur
            # after the version is aborted or committed.
            if sv:
                if sv != os:
                    self.client.invalidateVerify((oid, ''))
            else:
                if s != os:
                    self.client.invalidateVerify((oid, ''))

    def endZeoVerify(self):
        if self.verifying:
            self.stats.verifying_clients -= 1
        self.verifying = 0
        self.client.endVerify()

    def pack(self, time, wait=1):
        # Yes, you can pack a read-only server or storage!
        if wait:
            return run_in_thread(self._pack_impl, time)
        else:
            # If the client isn't waiting for a reply, start a thread
            # and forget about it.
            t = threading.Thread(target=self._pack_impl, args=(time,))
            t.start()
            return None

    def _pack_impl(self, time):
        self.log("pack(time=%s) started..." % repr(time))
        self.storage.pack(time, referencesf)
        self.log("pack(time=%s) complete" % repr(time))
        # Broadcast new size statistics
        self.server.invalidate(0, self.storage_id, None,
                               (), self.get_size_info())

    def new_oids(self, n=100):
        """Return a sequence of n new oids, where n defaults to 100"""
        if self.read_only:
            raise ReadOnlyError()
        if n <= 0:
            n = 1
        return [self.storage.new_oid() for i in range(n)]

    # undoLog and undoInfo are potentially slow methods

    def undoInfo(self, first, last, spec):
        return run_in_thread(self.storage.undoInfo, first, last, spec)

    def undoLog(self, first, last):
        return run_in_thread(self.storage.undoLog, first, last)

    def tpc_begin(self, id, user, description, ext, tid=None, status=" "):
        if self.read_only:
            raise ReadOnlyError()
        if self.transaction is not None:
            if self.transaction.id == id:
                self.log("duplicate tpc_begin(%s)" % repr(id))
                return
            else:
                raise StorageTransactionError("Multiple simultaneous tpc_begin"
                                              " requests from one client.")

        self.transaction = t = transaction.Transaction()
        t.id = id
        t.user = user
        t.description = description
        t._extension = ext

        self.serials = []
        self.invalidated = []
        self.txnlog = CommitLog()
        self.tid = tid
        self.status = status
        self.store_failed = 0
        self.stats.active_txns += 1

    def tpc_finish(self, id):
        if not self._check_tid(id):
            return
        assert self.locked
        self.stats.active_txns -= 1
        self.stats.commits += 1
        self.storage.tpc_finish(self.transaction)
        tid = self.storage.lastTransaction()
        if self.invalidated:
            self.server.invalidate(self, self.storage_id, tid,
                                   self.invalidated, self.get_size_info())
        self._clear_transaction()
        # Return the tid, for cache invalidation optimization
        return tid

    def tpc_abort(self, id):
        if not self._check_tid(id):
            return
        self.stats.active_txns -= 1
        self.stats.aborts += 1
        if self.locked:
            self.storage.tpc_abort(self.transaction)
        self._clear_transaction()

    def _clear_transaction(self):
        # Common code at end of tpc_finish() and tpc_abort()
        self.transaction = None
        self.txnlog.close()
        if self.locked:
            self.locked = 0
            self.timeout.end(self)
            self.stats.lock_time = None
            self.log("Transaction released storage lock", BLATHER)
            # _handle_waiting() can start another transaction (by
            # restarting a waiting one) so must be done last
            self._handle_waiting()

    def _abort(self):
        # called when a connection is closed unexpectedly
        if not self.locked:
            # Delete (d, zeo_storage) from the _waiting list, if found.
            waiting = self.storage._waiting
            for i in range(len(waiting)):
                d, z = waiting[i]
                if z is self:
                    del waiting[i]
                    self.log("Closed connection removed from waiting list."
                             " Clients waiting: %d." % len(waiting))
                    break

        if self.transaction:
            self.stats.active_txns -= 1
            self.stats.aborts += 1
            self.tpc_abort(self.transaction.id)

    # The public methods of the ZEO client API do not do the real work.
    # They defer work until after the storage lock has been acquired.
    # Most of the real implementations are in methods beginning with
    # an _.

    def storea(self, oid, serial, data, version, id):
        self._check_tid(id, exc=StorageTransactionError)
        self.stats.stores += 1
        self.txnlog.store(oid, serial, data, version)

    # The following four methods return values, so they must acquire
    # the storage lock and begin the transaction before returning.

    def vote(self, id):
        self._check_tid(id, exc=StorageTransactionError)
        if self.locked:
            return self._vote()
        else:
            return self._wait(lambda: self._vote())

    def abortVersion(self, src, id):
        self._check_tid(id, exc=StorageTransactionError)
        if self.locked:
            return self._abortVersion(src)
        else:
            return self._wait(lambda: self._abortVersion(src))

    def commitVersion(self, src, dest, id):
        self._check_tid(id, exc=StorageTransactionError)
        if self.locked:
            return self._commitVersion(src, dest)
        else:
            return self._wait(lambda: self._commitVersion(src, dest))

    def undo(self, trans_id, id):
        self._check_tid(id, exc=StorageTransactionError)
        if self.locked:
            return self._undo(trans_id)
        else:
            return self._wait(lambda: self._undo(trans_id))

    def _tpc_begin(self, txn, tid, status):
        self.locked = 1
        self.timeout.begin(self)
        self.stats.lock_time = time.time()
        self.storage.tpc_begin(txn, tid, status)

    def _store(self, oid, serial, data, version):
        err = None
        try:
            newserial = self.storage.store(oid, serial, data, version,
                                           self.transaction)
        except (SystemExit, KeyboardInterrupt):
            raise
        except Exception, err:
            self.store_failed = 1
            if isinstance(err, ConflictError):
                self.stats.conflicts += 1
                self.log("conflict error oid=%s msg=%s" %
                         (oid_repr(oid), str(err)), BLATHER)
            if not isinstance(err, TransactionError):
                # Unexpected errors are logged and passed to the client
                self.log("store error: %s, %s" % sys.exc_info()[:2],
                         logging.ERROR, exc_info=True)
            # Try to pickle the exception.  If it can't be pickled,
            # the RPC response would fail, so use something else.
            pickler = cPickle.Pickler()
            pickler.fast = 1
            try:
                pickler.dump(err, 1)
            except:
                msg = "Couldn't pickle storage exception: %s" % repr(err)
                self.log(msg, logging.ERROR)
                err = StorageServerError(msg)
            # The exception is reported back as newserial for this oid
            newserial = err
        else:
            if serial != "\0\0\0\0\0\0\0\0":
                self.invalidated.append((oid, version))
        if newserial == ResolvedSerial:
            self.stats.conflicts_resolved += 1
            self.log("conflict resolved oid=%s" % oid_repr(oid), BLATHER)
        self.serials.append((oid, newserial))
        return err is None

    def _vote(self):
        self.client.serialnos(self.serials)
        # If a store call failed, then return to the client immediately.
        # The serialnos() call will deliver an exception that will be
        # handled by the client in its tpc_vote() method.
        if self.store_failed:
            return
        return self.storage.tpc_vote(self.transaction)

    def _abortVersion(self, src):
        tid, oids = self.storage.abortVersion(src, self.transaction)
        inv = [(oid, src) for oid in oids]
        self.invalidated.extend(inv)
        return tid, oids

    def _commitVersion(self, src, dest):
        tid, oids = self.storage.commitVersion(src, dest, self.transaction)
        inv = [(oid, dest) for oid in oids]
        self.invalidated.extend(inv)
        if dest:
            inv = [(oid, src) for oid in oids]
            self.invalidated.extend(inv)
        return tid, oids

    def _undo(self, trans_id):
        tid, oids = self.storage.undo(trans_id, self.transaction)
        inv = [(oid, None) for oid in oids]
        self.invalidated.extend(inv)
        return tid, oids

    # When a delayed transaction is restarted, the dance is
    # complicated.  The restart occurs when one ZEOStorage instance
    # finishes as a transaction and finds another instance is in the
    # _waiting list.

    # It might be better to have a mechanism to explicitly send
    # the finishing transaction's reply before restarting the waiting
    # transaction.  If the restart takes a long time, the previous
    # client will be blocked until it finishes.

    def _wait(self, thunk):
        # Wait for the storage lock to be acquired.
        self._thunk = thunk
        if self.storage._transaction:
            d = Delay()
            self.storage._waiting.append((d, self))
            self.log("Transaction blocked waiting for storage. "
                     "Clients waiting: %d." % len(self.storage._waiting))
            return d
        else:
            self.log("Transaction acquired storage lock.", BLATHER)
            return self._restart()

    def _restart(self, delay=None):
        # Restart when the storage lock is available.
        if self.txnlog.stores == 1:
            template = "Preparing to commit transaction: %d object, %d bytes"
        else:
            template = "Preparing to commit transaction: %d objects, %d bytes"
        self.log(template % (self.txnlog.stores, self.txnlog.size()),
                 level=BLATHER)
        self._tpc_begin(self.transaction, self.tid, self.status)
        loads, loader = self.txnlog.get_loader()
        for i in range(loads):
            # load oid, serial, data, version
            if not self._store(*loader.load()):
                break
        resp = self._thunk()
        if delay is not None:
            delay.reply(resp)
        else:
            return resp

    def _handle_waiting(self):
        # Restart any client waiting for the storage lock.
        while self.storage._waiting:
            delay, zeo_storage = self.storage._waiting.pop(0)
            if self._restart_other(zeo_storage, delay):
                if self.storage._waiting:
                    n = len(self.storage._waiting)
                    self.log("Blocked transaction restarted.  "
                             "Clients waiting: %d" % n)
                else:
                    self.log("Blocked transaction restarted.")
                return

    def _restart_other(self, zeo_storage, delay):
        # Return True if the server restarted.
        # call the restart() method on the appropriate server.
        try:
            zeo_storage._restart(delay)
        except:
            self.log("Unexpected error handling waiting transaction",
                     level=logging.WARNING, exc_info=True)
            zeo_storage.connection.close()
            return 0
        else:
            return 1


class StorageServer:

    """The server side implementation of ZEO.

    The StorageServer is the 'manager' for incoming connections.  Each
    connection is associated with its own ZEOStorage instance (defined
    below).  The StorageServer may handle multiple storages; each
    ZEOStorage instance only handles a single storage.
    """

    # Classes we instantiate.  A subclass might override.

    DispatcherClass = Dispatcher
    ZEOStorageClass = ZEOStorage
    ManagedServerConnectionClass = ManagedServerConnection

    def __init__(self, addr, storages, read_only=0,
                 invalidation_queue_size=100,
                 transaction_timeout=None,
                 monitor_address=None,
                 auth_protocol=None,
                 auth_database=None,
                 auth_realm=None):
        """StorageServer constructor.

        This is typically invoked from the start.py script.

        Arguments (the first two are required and positional):

        addr -- the address at which the server should listen.  This
            can be a tuple (host, port) to signify a TCP/IP connection
            or a pathname string to signify a Unix domain socket
            connection.  A hostname may be a DNS name or a dotted IP
            address.

        storages -- a dictionary giving the storage(s) to handle.  The
            keys are the storage names, the values are the storage
            instances, typically FileStorage or Berkeley storage
            instances.  By convention, storage names are typically
            strings representing small integers starting at '1'.

        read_only -- an optional flag saying whether the server should
            operate in read-only mode.  Defaults to false.  Note that
            even if the server is operating in writable mode,
            individual storages may still be read-only.  But if the
            server is in read-only mode, no write operations are
            allowed, even if the storages are writable.  Note that
            pack() is considered a read-only operation.

        invalidation_queue_size -- The storage server keeps a queue
            of the objects modified by the last N transactions, where
            N == invalidation_queue_size.  This queue is used to
            speed client cache verification when a client disconnects
            for a short period of time.

        transaction_timeout -- The maximum amount of time to wait for
            a transaction to commit after acquiring the storage lock.
            If the transaction takes too long, the client connection
            will be closed and the transaction aborted.

        monitor_address -- The address at which the monitor server
            should listen.  If specified, a monitor server is started.
            The monitor server provides server statistics in a simple
            text format.

        auth_protocol -- The name of the authentication protocol to use.
            Examples are "digest" and "srp".

        auth_database -- The name of the password database filename.
            It should be in a format compatible with the authentication
            protocol used; for instance, "sha" and "srp" require different
            formats.

            Note that to implement an authentication protocol, a server
            and client authentication mechanism must be implemented in a
            auth_* module, which should be stored inside the "auth"
            subdirectory. This module may also define a DatabaseClass
            variable that should indicate what database should be used
            by the authenticator.
        """

        self.addr = addr
        self.storages = storages
        set_label()
        msg = ", ".join(
            ["%s:%s:%s" % (name, storage.isReadOnly() and "RO" or "RW",
                           storage.getName())
             for name, storage in storages.items()])
        log("%s created %s with storages: %s" %
            (self.__class__.__name__, read_only and "RO" or "RW", msg))
        for s in storages.values():
            s._waiting = []
        self.read_only = read_only
        self.auth_protocol = auth_protocol
        self.auth_database = auth_database
        self.auth_realm = auth_realm
        self.database = None
        if auth_protocol:
            self._setup_auth(auth_protocol)
        # A list of at most invalidation_queue_size invalidations.
        # The list is kept in sorted order with the most recent
        # invalidation at the front.  The list never has more than
        # self.invq_bound elements.
        self.invq = []
        self.invq_bound = invalidation_queue_size
        self.connections = {}
        self.dispatcher = self.DispatcherClass(addr,
                                               factory=self.new_connection)
        self.stats = {}
        self.timeouts = {}
        for name in self.storages.keys():
            self.stats[name] = StorageStats()
            if transaction_timeout is None:
                # An object with no-op methods
                timeout = StubTimeoutThread()
            else:
                timeout = TimeoutThread(transaction_timeout)
                timeout.start()
            self.timeouts[name] = timeout
        if monitor_address:
            self.monitor = StatsServer(monitor_address, self.stats)
        else:
            self.monitor = None

    def _setup_auth(self, protocol):
        # Can't be done in global scope, because of cyclic references
        from ZEO.auth import get_module

        name = self.__class__.__name__

        module = get_module(protocol)
        if not module:
            log("%s: no such an auth protocol: %s" % (name, protocol))
            return

        storage_class, client, db_class = module

        if not storage_class or not issubclass(storage_class, ZEOStorage):
            log(("%s: %s isn't a valid protocol, must have a StorageClass" %
                 (name, protocol)))
            self.auth_protocol = None
            return
        self.ZEOStorageClass = storage_class

        log("%s: using auth protocol: %s" % (name, protocol))

        # We create a Database instance here for use with the authenticator
        # modules. Having one instance allows it to be shared between multiple
        # storages, avoiding the need to bloat each with a new authenticator
        # Database that would contain the same info, and also avoiding any
        # possibly synchronization issues between them.
        self.database = db_class(self.auth_database)
        if self.database.realm != self.auth_realm:
            raise ValueError("password database realm %r "
                             "does not match storage realm %r"
                             % (self.database.realm, self.auth_realm))


    def new_connection(self, sock, addr):
        """Internal: factory to create a new connection.

        This is called by the Dispatcher class in ZEO.zrpc.server
        whenever accept() returns a socket for a new incoming
        connection.
        """
        if self.auth_protocol and self.database:
            zstorage = self.ZEOStorageClass(self, self.read_only,
                                            auth_realm=self.auth_realm)
            zstorage.set_database(self.database)
        else:
            zstorage = self.ZEOStorageClass(self, self.read_only)

        c = self.ManagedServerConnectionClass(sock, addr, zstorage, self)
        log("new connection %s: %s" % (addr, repr(c)))
        return c

    def register_connection(self, storage_id, conn):
        """Internal: register a connection with a particular storage.

        This is called by ZEOStorage.register().

        The dictionary self.connections maps each storage name to a
        list of current connections for that storage; this information
        is needed to handle invalidation.  This function updates this
        dictionary.

        Returns the timeout and stats objects for the appropriate storage.
        """
        l = self.connections.get(storage_id)
        if l is None:
            l = self.connections[storage_id] = []
        l.append(conn)
        stats = self.stats[storage_id]
        stats.clients += 1
        return self.timeouts[storage_id], stats

    def invalidate(self, conn, storage_id, tid, invalidated=(), info=None):
        """Internal: broadcast info and invalidations to clients.

        This is called from several ZEOStorage methods.

        This can do three different things:

        - If the invalidated argument is non-empty, it broadcasts
          invalidateTransaction() messages to all clients of the given
          storage except the current client (the conn argument).

        - If the invalidated argument is empty and the info argument
          is a non-empty dictionary, it broadcasts info() messages to
          all clients of the given storage, including the current
          client.

        - If both the invalidated argument and the info argument are
          non-empty, it broadcasts invalidateTransaction() messages to all
          clients except the current, and sends an info() message to
          the current client.

        """
        if invalidated:
            if len(self.invq) >= self.invq_bound:
                self.invq.pop()
            self.invq.insert(0, (tid, invalidated))
        for p in self.connections.get(storage_id, ()):
            if invalidated and p is not conn:
                p.client.invalidateTransaction(tid, invalidated)
            elif info is not None:
                p.client.info(info)

    def get_invalidations(self, tid):
        """Return a tid and list of all objects invalidation since tid.

        The tid is the most recent transaction id seen by the client.

        Returns None if it is unable to provide a complete list
        of invalidations for tid.  In this case, client should
        do full cache verification.
        """

        if not self.invq:
            log("invq empty")
            return None, []

        earliest_tid = self.invq[-1][0]
        if earliest_tid > tid:
            log("tid to old for invq %s < %s" % (u64(tid), u64(earliest_tid)))
            return None, []

        oids = {}
        for _tid, L in self.invq:
            if _tid <= tid:
                break
            for key in L:
                oids[key] = 1
        latest_tid = self.invq[0][0]
        return latest_tid, oids.keys()

    def close_server(self):
        """Close the dispatcher so that there are no new connections.

        This is only called from the test suite, AFAICT.
        """
        self.dispatcher.close()
        if self.monitor is not None:
            self.monitor.close()
        for storage in self.storages.values():
            storage.close()
        # Force the asyncore mainloop to exit by hackery, i.e. close
        # every socket in the map.  loop() will return when the map is
        # empty.
        for s in asyncore.socket_map.values():
            try:
                s.close()
            except:
                pass

    def close_conn(self, conn):
        """Internal: remove the given connection from self.connections.

        This is the inverse of register_connection().
        """
        for cl in self.connections.values():
            if conn.obj in cl:
                cl.remove(conn.obj)

class StubTimeoutThread:

    def begin(self, client):
        pass

    def end(self, client):
        pass

class TimeoutThread(threading.Thread):
    """Monitors transaction progress and generates timeouts."""

    # There is one TimeoutThread per storage, because there's one
    # transaction lock per storage.

    def __init__(self, timeout):
        threading.Thread.__init__(self)
        self.setDaemon(1)
        self._timeout = timeout
        self._client = None
        self._deadline = None
        self._cond = threading.Condition() # Protects _client and _deadline
        self._trigger = trigger()

    def begin(self, client):
        # Called from the restart code the "main" thread, whenever the
        # storage lock is being acquired.  (Serialized by asyncore.)
        self._cond.acquire()
        try:
            assert self._client is None
            self._client = client
            self._deadline = time.time() + self._timeout
            self._cond.notify()
        finally:
            self._cond.release()

    def end(self, client):
        # Called from the "main" thread whenever the storage lock is
        # being released.  (Serialized by asyncore.)
        self._cond.acquire()
        try:
            assert self._client is not None
            assert self._client is client
            self._client = None
            self._deadline = None
        finally:
            self._cond.release()

    def run(self):
        # Code running in the thread.
        while 1:
            self._cond.acquire()
            try:
                while self._deadline is None:
                    self._cond.wait()
                howlong = self._deadline - time.time()
                if howlong <= 0:
                    # Prevent reporting timeout more than once
                    self._deadline = None
                client = self._client # For the howlong <= 0 branch below
            finally:
                self._cond.release()
            if howlong <= 0:
                client.log("Transaction timeout after %s seconds" %
                           self._timeout)
                self._trigger.pull_trigger(lambda: client.connection.close())
            else:
                time.sleep(howlong)

def run_in_thread(method, *args):
    t = SlowMethodThread(method, args)
    t.start()
    return t.delay

class SlowMethodThread(threading.Thread):
    """Thread to run potentially slow storage methods.

    Clients can use the delay attribute to access the MTDelay object
    used to send a zrpc response at the right time.
    """

    # Some storage methods can take a long time to complete.  If we
    # run these methods via a standard asyncore read handler, they
    # will block all other server activity until they complete.  To
    # avoid blocking, we spawn a separate thread, return an MTDelay()
    # object, and have the thread reply() when it finishes.

    def __init__(self, method, args):
        threading.Thread.__init__(self)
        self._method = method
        self._args = args
        self.delay = MTDelay()

    def run(self):
        try:
            result = self._method(*self._args)
        except (SystemExit, KeyboardInterrupt):
            raise
        except Exception:
            self.delay.error(sys.exc_info())
        else:
            self.delay.reply(result)