zodb / relstorage / 1539

##############################################################################

#

# Copyright (c) 2019 Zope Foundation and Contributors.

# All Rights Reserved.

#

# This software is subject to the provisions of the Zope Public License,

# Version 2.1 (ZPL).  A copy of the ZPL should accompany this distribution.

# THIS SOFTWARE IS PROVIDED "AS IS" AND ANY AND ALL EXPRESS OR IMPLIED

# WARRANTIES ARE DISCLAIMED, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED

# WARRANTIES OF TITLE, MERCHANTABILITY, AGAINST INFRINGEMENT, AND FITNESS

# FOR A PARTICULAR PURPOSE.

#

##############################################################################

States related to TPC voting.

The implementation of locking the database for commit, flushing

temporary objects to the database, and moving them to their final locations,

live here.

"""

        # We're midway between the state of a database-wide lock

        # and consistent row-level locking. The lock here is now

        # a row-level artificial lock on COMMIT_ROW_LOCK, and we then

        # read TRANSACTION (or OBJECT_STATE in HF).

        # TODO: Continue working to remove the need for the artificial

        # lock.

            cursor,

            user,

            desc,

            ext

        )

                                    cursor, adapter, ude):

            cursor, tid_int, user, desc, ext, tid_is_packed)

        'tid',

        'tid_int',

        'release_commit_lock',

        'local_allocation_time',

    )

    def __repr__(self):

        return "<%s tid_int=%d created=%s>" %(

            self.__class__.__name__,

            self.tid_int,

            self.local_allocation_time

        )

    """

    The state we're in following ``tpc_vote``.

    Unlike the begin states, you *must* explicitly call :meth:`enter`

    on this object after it is constructed.

    """

        # (user, description, extension) from the transaction. byte objects.

        'ude',

        # required_tids: {oid_int: tid_int}; confirms that certain objects

        # have not changed at commit. May be a BTree

        'required_tids',

        # The DatabaseLockedForTid object

        'committing_tid_lock',

        # {oid_bytes}: Things that get changed as part of the vote process

        # and thus need to be invalidated. TODO: Move to shared state?

        'invalidated_oids',

        # How many conflicts there were to resolve. None if we're not there yet.

        'count_conflicts',

        # The timestamp we gained control after locking, and then the

        # timestamp we completed voting. If it takes "too long" to get

        # around to finishing, we'll log a warning.

        'lock_and_vote_times'

    )

        # type: (AbstractBegin, DatabaseLockedForTid) -> None

        # If committing_tid is passed to this method, it means the

        # database has already been locked and the TID is locked in.

        # This is (only!) done when we're restoring transactions.

        # Anything that we've undone or deleted is also invalidated.

            'share_lock_count': len(self.required_tids),

            'conflict_count': self.count_conflicts,

            'invalidated_count': len(self.invalidated_oids),

        }

            # Don't bother if we're empty.

        """

        Prepare the transaction for final commit.

        Locks (only!) the rows that will be updated or were marked as

        explicit dependencies through

        `checkCurrentSerialInTransaction`, and then verfies those

        dependencies and resolves conflicts.

        If we're using a shared blob dir, we then take out the commit

        lock, in order to move blobs into final position. (TODO: That

        might not be fully necessary, so long as we can properly roll

        back blobs.) Otherwise, this only locks the rows we will impact.

        """

        # It is assumed that self._lock.acquire was called before this

        # method was called.

        # execute all remaining batch store operations.

        # This exists as an extension point.

        # Reserve all OIDs used by this transaction.

        # In a typical transaction, this is our first write to the database

        # that's not temporary.

        # In the common case where we allocated OIDs for new objects,

        # the most recent OID we allocated will match the

        # ``max_stored_oid```. In the also-common case where we've

        # *never* allocated objects and we're just updating

        # pre-existing objects, every OID we see will be newer than

        # what we allocated (0, we never allocated). So it is the

        # responsibility of the ``relstorage.storage.oid.OIDs`` object

        # to keep track of seen OIDs across transacations, keeping

        # track of the maximum that it's seen, and only actually

        # sending a request down to the database when it's a genuinely

        # higher OID.

        #

        # The other time this can come up is ``copyTransactionsFrom()`` or ``restore()``:

        # those OIDs originated elsewhere and have no relation to our database sequence. That's

        # ok, the same logic applies.

        #

        # In this way, we minimize trips to the DB. We could be a

        # *little* bit smarter and track whether one of those APIs was

        # used, or whether we're updating existing objects and avoid a

        # bit more overhead, but benchmarking suggests that it's not

        # worth it in common cases.

            store_connection,

            self.shared_state.temp_storage.max_stored_oid)

        # Lock objects being modified and those registered with

        # readCurrent(). This could raise ReadConflictError or locking

        # errors. See ``IRelStorageAdapter`` for details.

        # Ok, we have now taken database locks: exclusive for each old

        # object we are updating and shared for each we wanted to

        # read-current. From now through ``tpc_finish`` we should do

        # our best to avoid anything that introduces unpredictable,

        # arbitrary latencies or delays. This is especially the case

        # if we're gevent monkey-patched: switching greenlets could

        # take an arbitrary amount of time to get back to *this*

        # greenlet so we can finish the job and release our locks.

        # That means we should limit opportunities that involve IO

        # (database queries) or even taking internal locks that may be

        # contended. We ask both our load and store connections to

        # boost their priority until commit so that any queries we do

        # make get serviced ASAP (some gevent drivers can actually

        # guarantee this), but only right before we're about to do something

        # that could potentially be unbounded or allow switching.

            # We've already picked a TID. Must have called undo().

            # Avoid accessing the actual blobhelper unless we need it

                # If this raises an STE, it must be a shared (non-db)

                # blobhelper, and the TID must not be locked.

            # It is crucial to do this only after locking the current

            # object rows in order to prevent deadlock. (The same order as a regular

            # transaction, just slightly sooner.)

        # New storage protocol

        """

        Either raises an `ConflictError`, or successfully resolves

        all conflicts.

        Returns a set of int OIDs for objects modified in this transaction

        but which were then updated by conflict resolution and so must

        be invalidated.

        All the rows needed for detecting conflicts should be locked against

        concurrent changes.

        :param conflicts: A sequence of information needed for detecting

           and resolving conflicts:

           ``(oid_int, committed_tid_int, tid_this_txn_saw_int, committed_state)``.

           If ``tid_this_txn_saw_int`` is None, it was a read-current check,

           and unless the ``committed_tid_int`` matches the expected value,

           a conflict error is raised.

        """

        # pylint:disable=too-many-locals

        # In the past, we didn't load all conflicts from the DB at

        # once, just one at a time. This was because we also fetched

        # the new state data from the DB, and it could be large (if

        # lots of conflicts). But now we use the state we have in our

        # local temp cache for the new state, so we don't need to

        # fetch it, meaning this result will be small...

        #

        # ...almost. The resolution process needs three pickles: the

        # one we tried to save, the one we're based off of, and the

        # one currently committed. Remember we have locked objects at

        # this point, so we need to finish ASAP to not block other

        # transactions; in gevent, we need to also avoid giving up

        # control to the event loop for arbitrary periods of time too

        # as it could take a long time to get back to us.

        # - The one we tried to save (the new one) is passed as a

        # parameter. We read this from our local storage, which is

        # probably in memory and thus fast.

        #

        # - The one currently committed can optionally be passed, and

        # if not, loadSerial() is used to get it. It seems somewhat

        # unlikely that it's not in the local pickle cache, so we

        # probably benefit from returning it in the conflict query.

        #

        # - The one we were based off of is always loaded with

        # loadSerial(). We *possibly* have the one we're based off of

        # already in our storage cache, but there's no guarantee. So

        # it's best to prefetch all these things in order to limit the

        # number of database round-trips and the opportunity to block

        # for arbitrary periods of time.

        # First, go through and distinguish read-current conflicts from

        # state conflicts (if the adapter didn't do that already).

                # An actual conflict. We need the state.

            else:

                # A readCurrent entry. Did it conflict?

                # Note that some database adapters (MySQL) may have already raised a

                # UnableToLockRowsToReadCurrentError indicating a conflict. That's a type

                # of ReadConflictError like this.

                        oid=int64_to_8bytes(oid_int),

                        serials=(int64_to_8bytes(committed_tid_int),

                                 int64_to_8bytes(expect_tid_int)))

            # Nothing to prefetch or resolve. No need to go critical,

            # we have no other opportunities to switch.

        # We're probably going to need to make a database query. Elevate our

        # priority and regain control ASAP.

            self.shared_state.load_connection.cursor,

            old_states_to_prefetch

        )

            storage, old_states_and_tids

        ).tryToResolveConflict

        # The conflicts can be very large binary strings, no need to include

        # them in traceback info. (Plus they could be sensitive.)

            # Match the names of the arguments used

            # Because we're using the _CachedConflictResolver, we can only loadSerial()

            # one state: the ``oldSerial`` state. Therefore the committedData *must* be

            # given.

                                                  newpickle, committedData)

                # unresolvable; kill the whole transaction

                    oid=oid,

                    serials=(oldSerial, committedSerial),

                    data=newpickle,

                )

            # resolved

        # We resolved some conflicts, so we need to send them over to the database.

            self.shared_state.store_connection.cursor,

            self.shared_state.temp_storage.iter_for_oids(invalidated_oid_ints)

        )

        # Here's where we take the global commit lock, and

        # allocate the next available transaction id, storing it

        # into history-preserving DBs. But if someone passed us

        # a TID (``restore`` or ``undo``), then it must already be in the DB, and the lock must

        # already be held.

        #

        # If we've prepared the transaction, then the TID must be in the

        # db, the lock must be held, and we must have finished all of our

        # storage actions. This is only expected to be the case when we have

        # a shared blob dir.

        #

        # Returns True if we also committed to the database.

            # Already done; *should* have been vote_only.

                                              self.committing_tid_lock)

            'commit': True

        }

            # Must be voting.

                # Avoid accessing the blobhelper unless we need it

           or self.shared_state.adapter.DEFAULT_LOCK_OBJECTS_AND_DETECT_CONFLICTS_INTERLEAVABLE:

            # If we're going to have to make two trips to the database, one to lock it and get a

            # tid and then one to commit and release locks, either because we're

            # just voting right now, not committing, or because the database doesn't

            # support doing that in a single operation, we need to go critical and

            # regain control ASAP so we can complete the operation.

        # Note that this may commit the load_connection and make it not

        # viable for a historical view anymore.

            self.shared_state.store_connection, self.shared_state.load_connection,

            self.shared_state.has_blobs(),

            self.ude,

            **kwargs

        )

            committing_tid_int, committing_tid_lock)

                int64_to_8bytes(committing_tid_int),

                committing_tid_int,

                self.shared_state.adapter

            )

                "tpc_finish called with wrong transaction")

            # Handle the finishing. We cannot/must not fail now.

            # TODO: Move most of this into the Finish class/module.

            # The IStorage docs say that f() "must be called while the

            # storage transaction lock is held." We don't really have a

            # "storage transaction lock", just the global database lock,

            # that we want to drop as quickly as possible, so it would be

            # nice to drop the commit lock and then call f(). This

            # probably doesn't really matter, though, as ZODB.Connection

            # doesn't use f().

            #

            # If we called `lock_and_move` for the first time in this

            # method, then the adapter will have been asked to go ahead

            # and commit, releasing any locks it can (some adapters do,

            # some don't). So we may or may not have a database lock at

            # this point.

                # Avoid accessing the actual blobhelper unless we need it

                    # If something failed to move, that's not really a problem:

                    # if we did any moving now, we're just a cache.

                        "Failed to update blob-cache; ignoring (will re-download)"

                    )

                "Objects were locked by %s for %.3fs",

                AbstractVote.tpc_finish.__wrapped__, # pylint:disable=no-member

                self, None,

                locked_duration,

                perf_logger

            )

                "Time between vote exiting and %s entering was %.3fs",

                AbstractVote.tpc_finish.__wrapped__, # pylint:disable=no-member

                self, None,

                between_vote_and_finish,

                perf_logger

            )

        finally:

        # Using undo() requires a new TID, so if we had already begun

        # a transaction by locking the database and allocating a TID,

        # we must preserve that.

                                                begin_state.committing_tid_lock)

        # We only get here if we've deleted objects, meaning we hold their row locks.

        # We only delete objects once we hold the commit lock.

        # Holding the commit lock put an entry in the transaction table,

        # but we don't want to bump the TID or store that data.

            self.shared_state.store_connection.cursor,

            self.committing_tid_lock.tid_int

        )

        # We don't do the final commit,

        # we just prepare.

            self.shared_state.store_connection,

            self.committing_tid_lock.tid_int

        )