ClientStorage.py 48.7 KB
Newer Older
1
##############################################################################
2
#
3
# Copyright (c) 2001, 2002, 2003 Zope Corporation and Contributors.
4
# All Rights Reserved.
5
#
6
# This software is subject to the provisions of the Zope Public License,
Jim Fulton's avatar
Jim Fulton committed
7
# Version 2.1 (ZPL).  A copy of the ZPL should accompany this distribution.
8 9 10 11
# 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
12
#
13
##############################################################################
14
"""The ClientStorage class and the exceptions that it may raise.
15

16 17 18
Public contents of this module:

ClientStorage -- the main class, implementing the Storage API
19
"""
Jim Fulton's avatar
Jim Fulton committed
20

Jeremy Hylton's avatar
Jeremy Hylton committed
21 22
import cPickle
import os
23
import socket
24
import sys
Jeremy Hylton's avatar
Jeremy Hylton committed
25 26 27
import tempfile
import threading
import time
28
import types
29
import logging
30

31
from zope.interface import implements
32 33
from ZEO import ServerStub
from ZEO.cache import ClientCache
Jeremy Hylton's avatar
Jeremy Hylton committed
34
from ZEO.TransactionBuffer import TransactionBuffer
Martijn Faassen's avatar
Martijn Faassen committed
35
from ZEO.Exceptions import ClientStorageError, ClientDisconnected, AuthError
36
from ZEO.auth import get_module
Jeremy Hylton's avatar
Jeremy Hylton committed
37
from ZEO.zrpc.client import ConnectionManager
38

39
import ZODB.lock_file
Jeremy Hylton's avatar
Jeremy Hylton committed
40
from ZODB import POSException
41
from ZODB import utils
42
from ZODB.loglevels import BLATHER
43
from ZODB.interfaces import IBlobStorage
44
from persistent.TimeStamp import TimeStamp
Jim Fulton's avatar
Jim Fulton committed
45

46 47 48 49 50 51
logger = logging.getLogger('ZEO.ClientStorage')
_pid = str(os.getpid())

def log2(msg, level=logging.INFO, subsys=_pid, exc_info=False):
    message = "(%s) %s" % (subsys, msg)
    logger.log(level, message, exc_info=exc_info)
52

53 54
try:
    from ZODB.ConflictResolution import ResolvedSerial
Jeremy Hylton's avatar
Jeremy Hylton committed
55 56
except ImportError:
    ResolvedSerial = 'rs'
Jim Fulton's avatar
Jim Fulton committed
57

58 59 60
def tid2time(tid):
    return str(TimeStamp(tid))

Jeremy Hylton's avatar
Jeremy Hylton committed
61
def get_timestamp(prev_ts=None):
62 63 64 65 66 67
    """Internal helper to return a unique TimeStamp instance.

    If the optional argument is not None, it must be a TimeStamp; the
    return value is then guaranteed to be at least 1 microsecond later
    the argument.
    """
Jeremy Hylton's avatar
Jeremy Hylton committed
68
    t = time.time()
69
    t = TimeStamp(*time.gmtime(t)[:5] + (t % 60,))
Jeremy Hylton's avatar
Jeremy Hylton committed
70 71 72
    if prev_ts is not None:
        t = t.laterThan(prev_ts)
    return t
73

Jeremy Hylton's avatar
Jeremy Hylton committed
74
class DisconnectedServerStub:
75 76 77 78 79 80 81
    """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, os it can be tested by identity.
    """
82

Jeremy Hylton's avatar
Jeremy Hylton committed
83 84
    def __getattr__(self, attr):
        raise ClientDisconnected()
85

86
# Singleton instance of DisconnectedServerStub
Jeremy Hylton's avatar
Jeremy Hylton committed
87
disconnected_stub = DisconnectedServerStub()
88

89 90
MB = 1024**2

91
class ClientStorage(object):
92

93 94 95 96 97 98 99 100
    """A Storage class that is a network client to a remote storage.

    This is a faithful implementation of the Storage API.

    This class is thread-safe; transactions are serialized in
    tpc_begin().
    """

101
    implements(IBlobStorage)
102 103 104
    # Classes we instantiate.  A subclass might override.

    TransactionBufferClass = TransactionBuffer
105
    ClientCacheClass = ClientCache
106 107 108
    ConnectionManagerClass = ConnectionManager
    StorageServerStubClass = ServerStub.StorageServer

109
    def __init__(self, addr, storage='1', cache_size=20 * MB,
110
                 name='', client=None, debug=0, var=None,
Jeremy Hylton's avatar
Jeremy Hylton committed
111
                 min_disconnect_poll=5, max_disconnect_poll=300,
112
                 wait_for_server_on_startup=None, # deprecated alias for wait
113
                 wait=None, wait_timeout=None,
114
                 read_only=0, read_only_fallback=0,
115
                 username='', password='', realm=None,
116
                 blob_dir=None, blob_cache_writable=False):
117 118 119 120 121 122 123 124
        """ClientStorage constructor.

        This is typically invoked from a custom_zodb.py file.

        All arguments except addr should be keyword arguments.
        Arguments:

        addr -- The server address(es).  This is either a list of
125 126
            addresses or a single address.  Each address can be a
            (hostname, port) tuple to signify a TCP/IP connection or
127 128 129 130
            a pathname string to signify a Unix domain socket
            connection.  A hostname may be a DNS name or a dotted IP
            address.  Required.

131
        storage -- The storage name, defaulting to '1'.  The name must
132
            match one of the storage names supported by the server(s)
133 134
            specified by the addr argument.  The storage name is
            displayed in the Zope control panel.
135 136 137 138 139 140 141

        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.

142 143
        client -- A name used to construct persistent cache filenames.
            Defaults to None, in which case the cache is not persistent.
144
            See ClientCache for more info.
145 146 147 148

        debug -- Ignored.  This is present only for backwards
            compatibility with ZEO 1.

149 150 151
        var -- When client is not None, this specifies the directory
            where the persistent cache files are created.  It defaults
            to None, in whichcase the current directory is used.
152 153 154 155 156 157 158 159 160 161 162 163 164 165 166

        min_disconnect_poll -- The minimum delay in seconds between
            attempts to connect to the server, in seconds.  Defaults
            to 5 seconds.

        max_disconnect_poll -- The maximum delay in seconds between
            attempts to connect to the server, in seconds.  Defaults
            to 300 seconds.

        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.

167 168 169
        wait_timeout -- Maximum time to wait for a connection before
            giving up.  Only meaningful if wait is True.

170 171 172 173 174 175 176 177 178
        read_only -- A flag indicating whether this should be a
            read-only storage, defaulting to false (i.e. writing is
            allowed by default).

        read_only_fallback -- A flag indicating whether a read-only
            remote storage should be acceptable as a fallback when no
            writable storages are available.  Defaults to false.  At
            most one of read_only and read_only_fallback should be
            true.
179 180 181 182

        username -- string with username to be used when authenticating.
            These only need to be provided if you are connecting to an
            authenticated server storage.
Tim Peters's avatar
Tim Peters committed
183

184 185 186
        password -- string with plaintext password to be used
            when authenticated.

187 188 189 190 191
        realm -- not documented.

        blob_dir -- directory path for blob data.  'blob data' is data that
            is retrieved via the loadBlob API.

192 193 194 195
        blob_cache_writable -- Flag whether the blob_dir is a writable shared
        filesystem that should be used instead of transferring blob data over
        zrpc.

196 197 198
        Note that the authentication protocol is defined by the server
        and is detected by the ClientStorage upon connecting (see
        testConnection() and doAuth() for details).
199 200
        """

201
        log2("%s (pid=%d) created %s/%s for storage: %r" %
202 203
             (self.__class__.__name__,
              os.getpid(),
204 205 206 207
              read_only and "RO" or "RW",
              read_only_fallback and "fallback" or "normal",
              storage))

208
        if debug:
209
            log2("ClientStorage(): debug argument is no longer used")
210 211 212 213 214

        # 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:
215 216 217
                log2("ClientStorage(): conflicting values for wait and "
                     "wait_for_server_on_startup; wait prevails",
                     level=logging.WARNING)
218
            else:
219
                log2("ClientStorage(): wait_for_server_on_startup "
220 221 222 223 224
                     "is deprecated; please use wait instead")
                wait = wait_for_server_on_startup
        elif wait is None:
            wait = 1

225
        self._addr = addr # For tests
226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243

        # 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.
Tim Peters's avatar
Tim Peters committed
244

Jeremy Hylton's avatar
Jeremy Hylton committed
245
        self._server = disconnected_stub
246 247 248
        self._connection = None
        self._pending_server = None
        self._ready = threading.Event()
249 250

        # _is_read_only stores the constructor argument
Jeremy Hylton's avatar
Jeremy Hylton committed
251
        self._is_read_only = read_only
252
        # _conn_is_read_only stores the status of the current connection
Jeremy Hylton's avatar
Jeremy Hylton committed
253
        self._conn_is_read_only = 0
Jeremy Hylton's avatar
Jeremy Hylton committed
254
        self._storage = storage
255
        self._read_only_fallback = read_only_fallback
256 257 258
        self._username = username
        self._password = password
        self._realm = realm
259 260 261 262 263

        # Flag tracking disconnections in the middle of a transaction.  This
        # is reset in tpc_begin() and set in notifyDisconnected().
        self._midtxn_disconnect = 0

264 265
        # _server_addr is used by sortKey()
        self._server_addr = None
266 267
        self._tfile = None
        self._pickler = None
268

Jeremy Hylton's avatar
Jeremy Hylton committed
269
        self._info = {'length': 0, 'size': 0, 'name': 'ZEO Client',
270
                      'supportsUndo':0, 'supportsVersions': 0}
271

272
        self._tbuf = self.TransactionBufferClass()
Jeremy Hylton's avatar
Jeremy Hylton committed
273
        self._db = None
274
        self._ltid = None # the last committed transaction
Guido van Rossum's avatar
Guido van Rossum committed
275

Jeremy Hylton's avatar
Jeremy Hylton committed
276 277 278
        # _serials: stores (oid, serialno) as returned by server
        # _seriald: _check_serials() moves from _serials to _seriald,
        #           which maps oid to serialno
279

280
        # TODO:  If serial number matches transaction id, then there is
281 282 283 284
        # 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.
Jeremy Hylton's avatar
Jeremy Hylton committed
285 286
        self._serials = []
        self._seriald = {}
287

Guido van Rossum's avatar
Guido van Rossum committed
288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305
        self.__name__ = name or str(addr) # Standard convention for storages

        # 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()

306 307 308 309 310 311 312 313 314
        # 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

315 316 317 318 319 320
        # 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()

321
        # XXX need to check for POSIX-ness here
322 323
        self.blob_dir = blob_dir
        self.blob_cache_writable = blob_cache_writable
324
        if blob_dir is not None:
325 326
            # Avoid doing this import unless we need it, as it
            # currently requires pywin32 on Windows.
327
            import ZODB.blob
328
            self.fshelper = ZODB.blob.FilesystemHelper(blob_dir)
329 330 331 332 333
            self.fshelper.create()
            self.fshelper.checkSecure()
        else:
            self.fshelper = None

Jeremy Hylton's avatar
Jeremy Hylton committed
334
        # Decide whether to use non-temporary files
335 336 337 338 339
        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
340
        self._cache = self.ClientCacheClass(cache_path, size=cache_size)
341
        # TODO:  maybe there's a better time to open the cache?  Unclear.
342
        self._cache.open()
Jeremy Hylton's avatar
Jeremy Hylton committed
343

344 345 346
        self._rpc_mgr = self.ConnectionManagerClass(addr, self,
                                                    tmin=min_disconnect_poll,
                                                    tmax=max_disconnect_poll)
Jeremy Hylton's avatar
Jeremy Hylton committed
347 348

        if wait:
349
            self._wait(wait_timeout)
Jeremy Hylton's avatar
Jeremy Hylton committed
350
        else:
351 352 353
            # 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.
Jeremy Hylton's avatar
Jeremy Hylton committed
354 355
            if not self._rpc_mgr.attempt_connect():
                self._rpc_mgr.connect()
356

357 358 359
    def _wait(self, timeout=None):
        if timeout is not None:
            deadline = time.time() + timeout
360
            log2("Setting deadline to %f" % deadline, level=BLATHER)
361 362
        else:
            deadline = None
363 364 365 366 367 368 369
        # 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.
370 371 372 373 374 375 376 377
        assert self._connection.is_async()
        while 1:
            self._ready.wait(30)
            if self._ready.isSet():
                break
            if timeout and time.time() > deadline:
                log2("Timed out waiting for connection",
                     level=logging.WARNING)
378
                break
379
            log2("Waiting for cache verification to finish")
380

Jeremy Hylton's avatar
Jeremy Hylton committed
381
    def close(self):
382
        """Storage API: finalize the storage, releasing external resources."""
383
        self._tbuf.close()
Jeremy Hylton's avatar
Jeremy Hylton committed
384 385
        if self._cache is not None:
            self._cache.close()
386 387 388 389
            self._cache = None
        if self._rpc_mgr is not None:
            self._rpc_mgr.close()
            self._rpc_mgr = None
390

391
    def registerDB(self, db):
392 393 394 395 396 397
        """Storage API: register a database for invalidation messages.

        This is called by ZODB.DB (and by some tests).

        The storage isn't really ready to use until after this call.
        """
Jeremy Hylton's avatar
Jeremy Hylton committed
398
        self._db = db
399

Jeremy Hylton's avatar
Jeremy Hylton committed
400
    def is_connected(self):
401
        """Return whether the storage is currently connected to a server."""
402 403 404
        # This function is used by clients, so we only report that a
        # connection exists when the connection is ready to use.
        return self._ready.isSet()
405

406
    def sync(self):
407 408
        # The separate async thread should keep us up to date
        pass
409

410 411
    def doAuth(self, protocol, stub):
        if not (self._username and self._password):
412
            raise AuthError("empty username or password")
413 414 415

        module = get_module(protocol)
        if not module:
416 417
            log2("%s: no such an auth protocol: %s" %
                 (self.__class__.__name__, protocol), level=logging.WARNING)
418 419 420 421 422
            return

        storage_class, client, db_class = module

        if not client:
423 424
            log2("%s: %s isn't a valid protocol, must have a Client class" %
                 (self.__class__.__name__, protocol), level=logging.WARNING)
425
            raise AuthError("invalid protocol")
Tim Peters's avatar
Tim Peters committed
426

427
        c = client(stub)
Tim Peters's avatar
Tim Peters committed
428

429 430
        # Initiate authentication, returns boolean specifying whether OK
        return c.start(self._username, self._realm, self._password)
Tim Peters's avatar
Tim Peters committed
431

432
    def testConnection(self, conn):
433
        """Internal: test the given connection.
434

435
        This returns:   1 if the connection is an optimal match,
436
                        0 if it is a suboptimal but acceptable match.
437 438
        It can also raise DisconnectedError or ReadOnlyError.

439 440 441 442 443 444 445 446 447 448 449 450 451
        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.
452
        """
453
        log2("Testing connection %r" % conn)
454
        # TODO:  Should we check the protocol version here?
Jeremy Hylton's avatar
Jeremy Hylton committed
455
        self._conn_is_read_only = 0
456
        stub = self.StorageServerStubClass(conn)
457 458

        auth = stub.getAuthProtocol()
459
        log2("Server authentication protocol %r" % auth)
460
        if auth:
461 462
            skey = self.doAuth(auth, stub)
            if skey:
463
                log2("Client authentication successful")
464
                conn.setSessionKey(skey)
465
            else:
466
                log2("Authentication failed")
467
                raise AuthError("Authentication failed")
Tim Peters's avatar
Tim Peters committed
468

469 470
        try:
            stub.register(str(self._storage), self._is_read_only)
471
            return 1
472
        except POSException.ReadOnlyError:
473
            if not self._read_only_fallback:
474
                raise
475
            log2("Got ReadOnlyError; trying again with read_only=1")
476
            stub.register(str(self._storage), read_only=1)
Jeremy Hylton's avatar
Jeremy Hylton committed
477
            self._conn_is_read_only = 1
478
            return 0
479

480
    def notifyConnected(self, conn):
481
        """Internal: start using the given connection.
482 483

        This is called by ConnectionManager after it has decided which
484
        connection should be used.
485
        """
486 487 488 489 490
        if self._cache is None:
            # the storage was closed, but the connect thread called
            # this method before it was stopped.
            return

491 492 493 494
        # invalidate our db cache
        if self._db is not None:
            self._db.invalidateCache()

495
        # TODO:  report whether we get a read-only connection.
496
        if self._connection is not None:
497
            reconnect = 1
498
        else:
499
            reconnect = 0
500
        self.set_server_addr(conn.get_addr())
501

502 503 504
        # 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.
505 506 507
        if self._connection is not None:
            self._connection.close()
        self._connection = conn
Jim Fulton's avatar
Jim Fulton committed
508

509
        if reconnect:
510
            log2("Reconnected to storage: %s" % self._server_addr)
511
        else:
512
            log2("Connected to storage: %s" % self._server_addr)
513

514 515 516
        stub = self.StorageServerStubClass(conn)
        self._oids = []
        self.verify_cache(stub)
517 518 519 520 521 522 523 524 525 526

        # 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())

        assert conn.is_async()

527
        self._handle_extensions()
528

529
    def _handle_extensions(self):
530
        for name in self.getExtensionMethods().keys():
531 532
            if not hasattr(self, name):
                setattr(self, name, self._server.extensionMethod(name))
533

534 535 536 537 538 539 540 541 542 543 544 545 546 547 548
    def set_server_addr(self, addr):
        # Normalize server address and convert to string
        if isinstance(addr, types.StringType):
            self._server_addr = addr
        else:
            assert isinstance(addr, types.TupleType)
            # If the server is on a remote host, we need to guarantee
            # that all clients used the same name for the server.  If
            # they don't, the sortKey() may be different for each client.
            # The best solution seems to be the official name reported
            # by gethostbyaddr().
            host = addr[0]
            try:
                canonical, aliases, addrs = socket.gethostbyaddr(host)
            except socket.error, err:
549 550
                log2("Error resolving host: %s (%s)" % (host, err),
                     level=BLATHER)
551 552 553 554 555 556 557 558 559
                canonical = host
            self._server_addr = str((canonical, addr[1]))

    def sortKey(self):
        # 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:
            raise ClientDisconnected
        else:
560
            return '%s:%s' % (self._storage, self._server_addr)
561

Jeremy Hylton's avatar
Jeremy Hylton committed
562
    def verify_cache(self, server):
563 564 565 566 567
        """Internal routine called to verify the cache.

        The return value (indicating which path we took) is used by
        the test suite.
        """
568 569 570 571

        # 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.
Tim Peters's avatar
Tim Peters committed
572

573 574 575 576
        last_inval_tid = self._cache.getLastTid()
        if last_inval_tid is not None:
            ltid = server.lastTransaction()
            if ltid == last_inval_tid:
577
                log2("No verification necessary (last_inval_tid up-to-date)")
578 579
                self._server = server
                self._ready.set()
580 581 582
                return "no verification"

            # log some hints about last transaction
583
            log2("last inval tid: %r %s\n"
584
                 % (last_inval_tid, tid2time(last_inval_tid)))
585
            log2("last transaction: %r %s" %
586 587 588 589
                 (ltid, ltid and tid2time(ltid)))

            pair = server.getInvalidations(last_inval_tid)
            if pair is not None:
590
                log2("Recovering %d invalidations" % len(pair[1]))
591
                self.invalidateTransaction(*pair)
592 593
                self._server = server
                self._ready.set()
594
                return "quick verification"
Tim Peters's avatar
Tim Peters committed
595

596
        log2("Verifying cache")
597 598 599 600 601
        # setup tempfile to hold zeoVerify results
        self._tfile = tempfile.TemporaryFile(suffix=".inv")
        self._pickler = cPickle.Pickler(self._tfile, 1)
        self._pickler.fast = 1 # Don't use the memo

602 603
        # TODO:  should batch these operations for efficiency; would need
        # to acquire lock ...
604 605
        for oid, tid, version in self._cache.contents():
            server.verify(oid, version, tid)
606
        self._pending_server = server
Jeremy Hylton's avatar
Jeremy Hylton committed
607
        server.endZeoVerify()
608
        return "full verification"
609 610 611 612 613 614 615 616 617

    ### 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.

Jeremy Hylton's avatar
Jeremy Hylton committed
618
    def notifyDisconnected(self):
619 620 621 622 623
        """Internal: notify that the server connection was terminated.

        This is called by ConnectionManager when the connection is
        closed or when certain problems with the connection occur.
        """
624
        log2("Disconnected from storage: %s" % repr(self._server_addr))
625
        self._connection = None
626
        self._ready.clear()
Jeremy Hylton's avatar
Jeremy Hylton committed
627
        self._server = disconnected_stub
628
        self._midtxn_disconnect = 1
629

Jeremy Hylton's avatar
Jeremy Hylton committed
630
    def __len__(self):
631
        """Return the size of the storage."""
632
        # TODO:  Is this method used?
Jeremy Hylton's avatar
Jeremy Hylton committed
633
        return self._info['length']
634

Jeremy Hylton's avatar
Jeremy Hylton committed
635
    def getName(self):
636 637 638 639 640 641 642 643 644 645 646
        """Storage API: return the storage name as a string.

        The return value consists of two parts: the name as determined
        by the name and addr argments to the ClientStorage
        constructor, and the string 'connected' or 'disconnected' in
        parentheses indicating whether the storage is (currently)
        connected.
        """
        return "%s (%s)" % (
            self.__name__,
            self.is_connected() and "connected" or "disconnected")
Jim Fulton's avatar
Jim Fulton committed
647

Jeremy Hylton's avatar
Jeremy Hylton committed
648
    def getSize(self):
649
        """Storage API: an approximate size of the database, in bytes."""
Jeremy Hylton's avatar
Jeremy Hylton committed
650
        return self._info['size']
651

652 653 654 655 656 657 658 659 660 661
    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
        """
662
        return self._info.get('extensionMethods', {})
663

Jeremy Hylton's avatar
Jeremy Hylton committed
664
    def supportsUndo(self):
665
        """Storage API: return whether we support undo."""
Jeremy Hylton's avatar
Jeremy Hylton committed
666
        return self._info['supportsUndo']
Jim Fulton's avatar
Jim Fulton committed
667

Jeremy Hylton's avatar
Jeremy Hylton committed
668
    def supportsVersions(self):
669
        """Storage API: return whether we support versions."""
Jeremy Hylton's avatar
Jeremy Hylton committed
670 671 672
        return self._info['supportsVersions']

    def isReadOnly(self):
673 674
        """Storage API: return whether we are in read-only mode."""
        if self._is_read_only:
Jeremy Hylton's avatar
Jeremy Hylton committed
675
            return 1
676 677 678 679 680
        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.
            return self._conn_is_read_only
Jeremy Hylton's avatar
Jeremy Hylton committed
681

682
    def _check_trans(self, trans):
683
        """Internal helper to check a transaction argument for sanity."""
684 685
        if self._is_read_only:
            raise POSException.ReadOnlyError()
Jeremy Hylton's avatar
Jeremy Hylton committed
686
        if self._transaction is not trans:
687 688
            raise POSException.StorageTransactionError(self._transaction,
                                                       trans)
Jim Fulton's avatar
Jim Fulton committed
689

690
    def abortVersion(self, version, txn):
691
        """Storage API: clear any changes made by the given version."""
692 693
        self._check_trans(txn)
        tid, oids = self._server.abortVersion(version, id(txn))
694 695 696 697 698 699 700
        # When a version aborts, invalidate the version and
        # non-version data.  The non-version data should still be
        # valid, but older versions of ZODB will change the
        # non-version serialno on an abort version.  With those
        # versions of ZODB, you'd get a conflict error if you tried to
        # commit a transaction with the cached data.

701
        # If we could guarantee that ZODB gave the right answer,
702
        # we could just invalidate the version data.
Jeremy Hylton's avatar
Jeremy Hylton committed
703
        for oid in oids:
704
            self._tbuf.invalidate(oid, '')
705
        return tid, oids
Jeremy Hylton's avatar
Jeremy Hylton committed
706

707
    def commitVersion(self, source, destination, txn):
708
        """Storage API: commit the source version in the destination."""
709 710
        self._check_trans(txn)
        tid, oids = self._server.commitVersion(source, destination, id(txn))
711
        if destination:
Jeremy Hylton's avatar
Jeremy Hylton committed
712 713
            # just invalidate our version data
            for oid in oids:
714
                self._tbuf.invalidate(oid, source)
Jeremy Hylton's avatar
Jeremy Hylton committed
715
        else:
716
            # destination is "", so invalidate version and non-version
Jeremy Hylton's avatar
Jeremy Hylton committed
717
            for oid in oids:
718 719
                self._tbuf.invalidate(oid, "")
        return tid, oids
720

721
    def history(self, oid, version, length=1):
722 723 724 725 726
        """Storage API: return a sequence of HistoryEntry objects.

        This does not support the optional filter argument defined by
        the Storage API.
        """
Jeremy Hylton's avatar
Jeremy Hylton committed
727 728
        return self._server.history(oid, version, length)

729
    def record_iternext(self, next=None):
730
        """Storage API: get the next database record.
731 732 733 734 735

        This is part of the conversion-support API.
        """
        return self._server.record_iternext(next)

736
    def getTid(self, oid):
737
        """Storage API: return current serial number for oid."""
738
        return self._server.getTid(oid)
739

740
    def loadSerial(self, oid, serial):
741
        """Storage API: load a historical revision of an object."""
Jeremy Hylton's avatar
Jeremy Hylton committed
742
        return self._server.loadSerial(oid, serial)
Jim Fulton's avatar
Jim Fulton committed
743

744 745 746 747 748 749 750
    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 and version, if they exist;
        otherwise a KeyError is raised.
        """
751 752 753
        return self.loadEx(oid, version)[:2]

    def loadEx(self, oid, version):
754 755
        self._lock.acquire()    # for atomic processing of invalidations
        try:
756 757 758
            t = self._cache.load(oid, version)
            if t:
                return t
759 760
        finally:
            self._lock.release()
761

Jeremy Hylton's avatar
Jeremy Hylton committed
762 763
        if self._server is None:
            raise ClientDisconnected()
764 765 766 767 768 769 770 771 772 773

        self._load_lock.acquire()
        try:
            self._lock.acquire()
            try:
                self._load_oid = oid
                self._load_status = 1
            finally:
                self._lock.release()

774
            data, tid, ver = self._server.loadEx(oid, version)
775 776 777 778

            self._lock.acquire()    # for atomic processing of invalidations
            try:
                if self._load_status:
779
                    self._cache.store(oid, ver, tid, None, data)
780 781 782 783 784 785
                self._load_oid = None
            finally:
                self._lock.release()
        finally:
            self._load_lock.release()

786 787 788 789 790 791 792 793 794 795 796 797 798 799 800 801 802 803 804 805
        return data, tid, ver

    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.

806
            # If we call again, we're guaranteed to get the
807 808 809 810 811 812 813 814 815 816 817 818 819 820 821 822
            # 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
Jeremy Hylton's avatar
Jeremy Hylton committed
823

Jim Fulton's avatar
Jim Fulton committed
824
    def modifiedInVersion(self, oid):
825 826 827 828
        """Storage API: return the version, if any, that modfied an object.

        If no version modified the object, return an empty string.
        """
829 830 831 832 833 834 835
        self._lock.acquire()
        try:
            v = self._cache.modifiedInVersion(oid)
            if v is not None:
                return v
        finally:
            self._lock.release()
Jeremy Hylton's avatar
Jeremy Hylton committed
836
        return self._server.modifiedInVersion(oid)
Jim Fulton's avatar
Jim Fulton committed
837

838 839
    def new_oid(self):
        """Storage API: return a new object identifier."""
Jeremy Hylton's avatar
Jeremy Hylton committed
840 841 842
        if self._is_read_only:
            raise POSException.ReadOnlyError()
        # avoid multiple oid requests to server at the same time
843
        self._oid_lock.acquire()
844 845 846 847 848 849 850
        try:
            if not self._oids:
                self._oids = self._server.new_oids()
                self._oids.reverse()
            return self._oids.pop()
        finally:
            self._oid_lock.release()
Jeremy Hylton's avatar
Jeremy Hylton committed
851

852 853 854 855 856 857 858 859 860 861 862 863 864
    def pack(self, t=None, referencesf=None, wait=1, days=0):
        """Storage API: pack the storage.

        Deviations from the Storage API: the referencesf argument is
        ignored; two additional optional arguments wait and days are
        provided:

        wait -- a flag indicating whether to wait for the pack to
            complete; defaults to true.

        days -- a number of days to subtract from the pack time;
            defaults to zero.
        """
865 866
        # TODO: Is it okay that read-only connections allow pack()?
        # rf argument ignored; server will provide its own implementation
Jeremy Hylton's avatar
Jeremy Hylton committed
867 868 869 870 871 872
        if t is None:
            t = time.time()
        t = t - (days * 86400)
        return self._server.pack(t, wait)

    def _check_serials(self):
873
        """Internal helper to move data from _serials to _seriald."""
874
        # serials are always going to be the same, the only
875
        # question is whether an exception has been raised.
Jeremy Hylton's avatar
Jeremy Hylton committed
876 877 878 879 880 881 882 883 884
        if self._serials:
            l = len(self._serials)
            r = self._serials[:l]
            del self._serials[:l]
            for oid, s in r:
                if isinstance(s, Exception):
                    raise s
                self._seriald[oid] = s
            return r
Jim Fulton's avatar
Jim Fulton committed
885

886
    def store(self, oid, serial, data, version, txn):
887
        """Storage API: store data for an object."""
888 889
        self._check_trans(txn)
        self._server.storea(oid, serial, data, version, id(txn))
Jeremy Hylton's avatar
Jeremy Hylton committed
890 891
        self._tbuf.store(oid, version, data)
        return self._check_serials()
Jim Fulton's avatar
Jim Fulton committed
892

893 894 895
    def storeBlob(self, oid, serial, data, blobfilename, version, txn):
        """Storage API: store a blob object."""
        serials = self.store(oid, serial, data, version, txn)
896
        if self.blob_cache_writable:
897 898
            self._storeBlob_shared(
                oid, serial, data, blobfilename, version, txn)
899
        else:
900 901 902 903
            self._server.storeBlob(
                oid, serial, data, blobfilename, version, txn)
            if blobfilename is not None:
                self._tbuf.storeBlob(oid, blobfilename)
904 905 906 907 908 909 910 911 912
        return serials

    def _storeBlob_shared(self, oid, serial, data, filename, version, txn):
        # First, move the blob into the blob directory
        dir = self.fshelper.getPathForOID(oid)
        if not os.path.exists(dir):
            os.mkdir(dir)
        fd, target = self.fshelper.blob_mkstemp(oid, serial)
        os.close(fd)
913

914 915 916 917 918 919 920 921 922 923 924 925 926 927 928 929 930 931 932 933 934
        if sys.platform == 'win32':

            # On windows, we can't rename to an existing file.  That's
            # OK.  We don't care what file we get as long as it is
            # unique.  We'll just keep trying until the rename suceeds.
            os.remove(target)
            i = 0
            while 1:
                try:
                    os.rename(filename, target + str(i))
                except OSError:
                    i += 1
                else:
                    break
            target += str(i)
        else:
            os.rename(filename, target)
        # Now tell the server where we put it
        self._server.storeBlobShared(
            oid, serial, data,
            os.path.basename(target), version, id(txn))
935

936 937 938 939 940 941
    def _have_blob(self, blob_filename, oid, serial):
        if os.path.exists(blob_filename):
            log2("Found blob %s/%s in cache." % (utils.oid_repr(oid),
                utils.tid_repr(serial)), level=BLATHER)
            return True
        return False
942

Christian Theune's avatar
Christian Theune committed
943
    def receiveBlobStart(self, oid, serial):
944 945 946 947 948 949 950 951
        blob_filename = self.fshelper.getBlobFilename(oid, serial)
        assert not os.path.exists(blob_filename)
        assert os.path.exists(blob_filename+'.lock')
        blob_filename += '.dl'
        assert not os.path.exists(blob_filename)
        f = open(blob_filename, 'wb')
        f.close()

Christian Theune's avatar
Christian Theune committed
952
    def receiveBlobChunk(self, oid, serial, chunk):
953 954 955 956 957
        blob_filename = self.fshelper.getBlobFilename(oid, serial)+'.dl'
        assert os.path.exists(blob_filename)
        f = open(blob_filename, 'ab')
        f.write(chunk)
        f.close()
958

Christian Theune's avatar
Christian Theune committed
959
    def receiveBlobStop(self, oid, serial):
960 961
        blob_filename = self.fshelper.getBlobFilename(oid, serial)
        os.rename(blob_filename+'.dl', blob_filename)
962

963
    def loadBlob(self, oid, serial):
964

965 966 967
        # Load a blob.  If it isn't present and we have a shared blob
        # directory, then assume that it doesn't exist on the server
        # and return None.
968 969 970 971 972 973
        if self.fshelper is None:
            raise POSException.Unsupported("No blob cache directory is "
                                           "configured.")

        blob_filename = self.fshelper.getBlobFilename(oid, serial)
        # Case 1: Blob is available already, just use it
974
        if self._have_blob(blob_filename, oid, serial):
975 976
            return blob_filename

977 978 979 980
        if self.blob_cache_writable:
            # We're using a server shared cache.  If the file isn't
            # here, it's not anywahere.
            return None
981

982 983 984 985 986 987 988 989 990 991 992 993 994 995 996 997
        # First, we'll create the directory for this oid, if it doesn't exist. 
        targetpath = self.fshelper.getPathForOID(oid)
        if not os.path.exists(targetpath):
            try:
                os.makedirs(targetpath, 0700)
            except OSError:
                # We might have lost a race.  If so, the directory
                # must exist now
                assert os.path.exists(targetpath)

        # OK, it's not here and we (or someone) needs to get it.  We
        # want to avoid getting it multiple times.  We want to avoid
        # getting it multiple times even accross separate client
        # processes on the same machine. We'll use file locking.

        lockfilename = blob_filename+'.lock'
998
        try:
999 1000 1001 1002 1003 1004 1005 1006 1007 1008 1009 1010 1011 1012 1013 1014 1015 1016 1017 1018 1019
            lock = ZODB.lock_file.LockFile(lockfilename)
        except ZODB.lock_file.LockError:

            # Someone is already downloading the Blob. Wait for the
            # lock to be freed.  How long should we be willing to wait?
            # TODO: maybe find some way to assess download progress.

            while 1:
                time.sleep(0.1)
                try:
                    lock = ZODB.lock_file.LockFile(lockfilename)
                except ZODB.lock_file.LockError:
                    pass
                else:
                    # We have the lock. We should be able to get the file now.
                    lock.close()
                    try:
                        os.remove(lockfilename)
                    except OSError:
                        pass
                    break
1020

1021 1022 1023 1024
            if self._have_blob(blob_filename, oid, serial):
                return blob_filename

            return None
1025 1026

        try:
1027 1028 1029 1030 1031
            # We got the lock, so it's our job to download it.  First,
            # we'll double check that someone didn't download it while we
            # were getting the lock:

            if self._have_blob(blob_filename, oid, serial):
1032 1033
                return blob_filename

1034 1035 1036
            # Ask the server to send it to us.  When this function
            # returns, it will have been sent. (The recieving will
            # have been handled by the asyncore thread.)
1037

1038
            self._server.sendBlob(oid, serial)
1039

1040 1041 1042 1043
            if self._have_blob(blob_filename, oid, serial):
                return blob_filename

            return None
1044

1045 1046 1047 1048 1049 1050
        finally:
            lock.close()
            try:
                os.remove(lockfilename)
            except OSError:
                pass
1051

1052 1053 1054
    def temporaryDirectory(self):
        return self.blob_dir

1055
    def tpc_vote(self, txn):
1056
        """Storage API: vote on a transaction."""
1057
        if txn is not self._transaction:
Jeremy Hylton's avatar
Jeremy Hylton committed
1058
            return
1059
        self._server.vote(id(txn))
Jeremy Hylton's avatar
Jeremy Hylton committed
1060 1061
        return self._check_serials()

1062 1063 1064
    def tpc_transaction(self):
        return self._transaction

Jeremy Hylton's avatar
Jeremy Hylton committed
1065
    def tpc_begin(self, txn, tid=None, status=' '):
1066
        """Storage API: begin a transaction."""
1067 1068
        if self._is_read_only:
            raise POSException.ReadOnlyError()
1069
        self._tpc_cond.acquire()
1070
        self._midtxn_disconnect = 0
Jeremy Hylton's avatar
Jeremy Hylton committed
1071
        while self._transaction is not None:
1072 1073 1074
            # 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.
Jeremy Hylton's avatar
Jeremy Hylton committed
1075
            if self._transaction == txn:
1076
                self._tpc_cond.release()
Jeremy Hylton's avatar
Jeremy Hylton committed
1077
                return
1078
            self._tpc_cond.wait(30)
Jeremy Hylton's avatar
Jeremy Hylton committed
1079
        self._transaction = txn
1080
        self._tpc_cond.release()
1081 1082

        try:
1083
            self._server.tpc_begin(id(txn), txn.user, txn.description,
Jeremy Hylton's avatar
Jeremy Hylton committed
1084
                                   txn._extension, tid, status)
Jeremy Hylton's avatar
Jeremy Hylton committed
1085 1086 1087
        except:
            # Client may have disconnected during the tpc_begin().
            if self._server is not disconnected_stub:
1088
                self.end_transaction()
Jeremy Hylton's avatar
Jeremy Hylton committed
1089 1090
            raise

1091
        self._tbuf.clear()
Jeremy Hylton's avatar
Jeremy Hylton committed
1092 1093
        self._seriald.clear()
        del self._serials[:]
Jim Fulton's avatar
Jim Fulton committed
1094

1095
    def end_transaction(self):
1096
        """Internal helper to end a transaction."""
1097
        # the right way to set self._transaction to None
1098 1099
        # calls notify() on _tpc_cond in case there are waiting threads
        self._tpc_cond.acquire()
1100
        self._transaction = None
1101 1102
        self._tpc_cond.notify()
        self._tpc_cond.release()
1103

1104
    def lastTransaction(self):
1105
        return self._cache.getLastTid()
1106

1107
    def tpc_abort(self, txn):
1108
        """Storage API: abort a transaction."""
1109
        if txn is not self._transaction:
1110 1111
            return
        try:
1112
            # Caution:  Are there any exceptions that should prevent an
1113 1114 1115 1116
            # abort from occurring?  It seems wrong to swallow them
            # all, yet you want to be sure that other abort logic is
            # executed regardless.
            try:
1117
                self._server.tpc_abort(id(txn))
1118
            except ClientDisconnected:
1119 1120
                log2("ClientDisconnected in tpc_abort() ignored",
                     level=BLATHER)
1121
        finally:
1122 1123 1124
            self._tbuf.clear()
            self._seriald.clear()
            del self._serials[:]
1125
            self.end_transaction()
1126

1127
    def tpc_finish(self, txn, f=None):
1128
        """Storage API: finish a transaction."""
1129
        if txn is not self._transaction:
Jeremy Hylton's avatar
Jeremy Hylton committed
1130
            return
1131
        self._load_lock.acquire()
1132
        try:
1133 1134 1135 1136
            if self._midtxn_disconnect:
                raise ClientDisconnected(
                       'Calling tpc_finish() on a disconnected transaction')

1137 1138 1139 1140
            # The calls to tpc_finish() and _update_cache() should
            # never run currently with another thread, because the
            # tpc_cond condition variable prevents more than one
            # thread from calling tpc_finish() at a time.
1141
            tid = self._server.tpc_finish(id(txn))
1142 1143
            self._lock.acquire()  # for atomic processing of invalidations
            try:
1144
                self._update_cache(tid)
1145
                if f is not None:
1146
                    f(tid)
1147 1148
            finally:
                self._lock.release()
Jeremy Hylton's avatar
Jeremy Hylton committed
1149 1150 1151 1152

            r = self._check_serials()
            assert r is None or len(r) == 0, "unhandled serialnos: %s" % r
        finally:
1153
            self._load_lock.release()
1154
            self.end_transaction()
Jeremy Hylton's avatar
Jeremy Hylton committed
1155

1156
    def _update_cache(self, tid):
1157 1158 1159 1160 1161
        """Internal helper to handle objects modified by a transaction.

        This iterates over the objects in the transaction buffer and
        update or invalidate the cache.
        """
1162
        # Must be called with _lock already acquired.
1163

1164
        # Not sure why _update_cache() would be called on a closed storage.
1165 1166 1167
        if self._cache is None:
            return

1168
        for oid, version, data in self._tbuf:
1169 1170 1171
            self._cache.invalidate(oid, version, tid)
            # If data is None, we just invalidate.
            if data is not None:
Jeremy Hylton's avatar
Jeremy Hylton committed
1172
                s = self._seriald[oid]
1173 1174 1175
                if s != ResolvedSerial:
                    assert s == tid, (s, tid)
                    self._cache.store(oid, version, s, None, data)
1176 1177 1178 1179 1180 1181 1182 1183 1184 1185 1186 1187 1188 1189

        
        if self.fshelper is not None:
            blobs = self._tbuf.blobs
            while blobs:
                oid, blobfilename = blobs.pop()
                targetpath = self.fshelper.getPathForOID(oid)
                if not os.path.exists(targetpath):
                    os.makedirs(targetpath, 0700)
                os.rename(blobfilename,
                          self.fshelper.getBlobFilename(oid, tid),
                          )

                    
Jeremy Hylton's avatar
Jeremy Hylton committed
1190
        self._tbuf.clear()
1191

1192
    def undo(self, trans_id, txn):
1193 1194 1195 1196 1197 1198 1199 1200
        """Storage API: undo a transaction.

        This is executed in a transactional context.  It has no effect
        until the transaction is committed.  It can be undone itself.

        Zope uses this to implement undo unless it is not supported by
        a storage.
        """
1201
        self._check_trans(txn)
1202
        tid, oids = self._server.undo(trans_id, id(txn))
Jeremy Hylton's avatar
Jeremy Hylton committed
1203 1204
        for oid in oids:
            self._tbuf.invalidate(oid, '')
1205
        return tid, oids
Jim Fulton's avatar
Jim Fulton committed
1206

1207
    def undoInfo(self, first=0, last=-20, specification=None):
1208
        """Storage API: return undo information."""
Jeremy Hylton's avatar
Jeremy Hylton committed
1209
        return self._server.undoInfo(first, last, specification)
Jim Fulton's avatar
Jim Fulton committed
1210

1211
    def undoLog(self, first=0, last=-20, filter=None):
1212
        """Storage API: return a sequence of TransactionDescription objects.
Jeremy Hylton's avatar
Jeremy Hylton committed
1213

1214 1215 1216 1217 1218 1219 1220 1221
        The filter argument should be None or left unspecified, since
        it is impossible to pass the filter function to the server to
        be executed there.  If filter is not None, an empty sequence
        is returned.
        """
        if filter is not None:
            return []
        return self._server.undoLog(first, last)
Jim Fulton's avatar
Jim Fulton committed
1222 1223

    def versionEmpty(self, version):
1224
        """Storage API: return whether the version has no transactions."""
Jeremy Hylton's avatar
Jeremy Hylton committed
1225
        return self._server.versionEmpty(version)
Jim Fulton's avatar
Jim Fulton committed
1226 1227

    def versions(self, max=None):
1228
        """Storage API: return a sequence of versions in the storage."""
Jeremy Hylton's avatar
Jeremy Hylton committed
1229 1230
        return self._server.versions(max)

1231
    # Below are methods invoked by the StorageServer
Jeremy Hylton's avatar
Jeremy Hylton committed
1232 1233

    def serialnos(self, args):
1234
        """Server callback to pass a list of changed (oid, serial) pairs."""
Jeremy Hylton's avatar
Jeremy Hylton committed
1235 1236 1237
        self._serials.extend(args)

    def info(self, dict):
1238
        """Server callback to update the info dictionary."""
Jeremy Hylton's avatar
Jeremy Hylton committed
1239 1240
        self._info.update(dict)

1241
    def invalidateVerify(self, args):
1242 1243 1244 1245
        """Server callback to invalidate an (oid, version) pair.

        This is called as part of cache validation.
        """
1246 1247
        # Invalidation as result of verify_cache().
        # Queue an invalidate for the end the verification procedure.
Jeremy Hylton's avatar
Jeremy Hylton committed
1248
        if self._pickler is None:
1249 1250
            # This should never happen.  TODO:  assert it doesn't, or log
            # if it does.
Jeremy Hylton's avatar
Jeremy Hylton committed
1251 1252 1253
            return
        self._pickler.dump(args)

1254 1255 1256 1257
    def _process_invalidations(self, invs):
        # Invalidations are sent by the ZEO server as a sequence of
        # oid, version pairs.  The DB's invalidate() method expects a
        # dictionary of oids.
1258

1259 1260 1261 1262
        self._lock.acquire()
        try:
            # versions maps version names to dictionary of invalidations
            versions = {}
1263
            for oid, version, tid in invs:
1264 1265
                if oid == self._load_oid:
                    self._load_status = 0
1266
                self._cache.invalidate(oid, version, tid)
1267 1268 1269 1270 1271
                oids = versions.get((version, tid))
                if not oids:
                    versions[(version, tid)] = [oid]
                else:
                    oids.append(oid)
1272

1273
            if self._db is not None:
1274 1275
                for (version, tid), d in versions.items():
                    self._db.invalidate(tid, d, version=version)
1276 1277
        finally:
            self._lock.release()
1278

1279
    def endVerify(self):
1280
        """Server callback to signal end of cache validation."""
Jeremy Hylton's avatar
Jeremy Hylton committed
1281 1282
        if self._pickler is None:
            return
1283 1284
        # write end-of-data marker
        self._pickler.dump((None, None))
1285
        self._pickler = None
Jeremy Hylton's avatar
Jeremy Hylton committed
1286 1287 1288
        self._tfile.seek(0)
        f = self._tfile
        self._tfile = None
1289
        self._process_invalidations(InvalidationLogIterator(f))
Jeremy Hylton's avatar
Jeremy Hylton committed
1290 1291
        f.close()

1292
        log2("endVerify finishing")
1293 1294 1295
        self._server = self._pending_server
        self._ready.set()
        self._pending_conn = None
1296
        log2("endVerify finished")
1297

1298 1299
    def invalidateTransaction(self, tid, args):
        """Invalidate objects modified by tid."""
1300 1301 1302 1303 1304
        self._lock.acquire()
        try:
            self._cache.setLastTid(tid)
        finally:
            self._lock.release()
1305
        if self._pickler is not None:
1306 1307
            log2("Transactional invalidation during cache verification",
                 level=BLATHER)
1308
            for t in args:
1309
                self._pickler.dump(t)
1310
            return
1311 1312
        self._process_invalidations([(oid, version, tid)
                                     for oid, version in args])
1313 1314 1315 1316 1317 1318

    # The following are for compatibility with protocol version 2.0.0

    def invalidateTrans(self, args):
        return self.invalidateTransaction(None, args)

1319 1320 1321
    invalidate = invalidateVerify
    end = endVerify
    Invalidate = invalidateTrans
1322

1323 1324 1325 1326
def InvalidationLogIterator(fileobj):
    unpickler = cPickle.Unpickler(fileobj)
    while 1:
        oid, version = unpickler.load()
1327
        if oid is None:
1328 1329
            break
        yield oid, version, None