18443719934

Committed 01 Oct 2025 07:43AM UTC coverage: 83.781% (-0.01%) from 83.793%

Build # 18443719934

Build Type

push

github

Committed by

dataflake

Commit Message

- vb [ci skip]

Run Details

2441 of 3542 branches covered (68.92%)

13353 of 15938 relevant lines covered (83.78%)

0.84 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

97.22

/src/ZODB/tests/MVCCMappingStorage.py

##############################################################################
#
# Copyright (c) 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
#
##############################################################################
"""An extension of MappingStorage that depends on polling.

Each Connection has its own view of the database.  Polling updates each
connection's view.
"""

from zope.interface import implementer

import ZODB.POSException
import ZODB.utils
from ZODB.interfaces import IMVCCStorage
from ZODB.MappingStorage import MappingStorage


@implementer(IMVCCStorage)
class MVCCMappingStorage(MappingStorage):

    def __init__(self, name="MVCC Mapping Storage"):
        MappingStorage.__init__(self, name=name)
        # _polled_tid contains the transaction ID at the last poll.
        self._polled_tid = b''
        self._data_snapshot = None  # {oid->(state, tid)}
        self._main_lock = self._lock

    def new_instance(self):
        """Returns a storage instance that is a view of the same data.
        """
        inst = MVCCMappingStorage(name=self.__name__)
        # All instances share the same OID data, transaction log, commit lock,
        # and OID sequence.
        inst._data = self._data
        inst._transactions = self._transactions
        inst._commit_lock = self._commit_lock
        inst.new_oid = self.new_oid
        inst.pack = self.pack
        inst.loadBefore = self.loadBefore
        inst._ltid = self._ltid
        inst._main_lock = self._lock
        return inst

    @ZODB.utils.locked(MappingStorage.opened)
    def sync(self, force=False):
        self._data_snapshot = None

    def release(self):
        pass

    @ZODB.utils.locked(MappingStorage.opened)
    def load(self, oid, version=''):
        assert not version, "Versions are not supported"
        if self._data_snapshot is None:
            self.poll_invalidations()
        info = self._data_snapshot.get(oid)
        if info:
            return info
        raise ZODB.POSException.POSKeyError(oid)

    def poll_invalidations(self):
        """Poll the storage for changes by other connections.
        """
        # prevent changes to _transactions and _data during analysis
        with self._main_lock:
            if self._transactions:
                new_tid = self._transactions.maxKey()
            else:
                new_tid = ZODB.utils.z64

            # Copy the current data into a snapshot. This is obviously
            # very inefficient for large storages, but it's good for
            # tests.
            self._data_snapshot = {}
            for oid, tid_data in self._data.items():
                if tid_data:
                    tid = tid_data.maxKey()
                    self._data_snapshot[oid] = tid_data[tid], tid

            if self._polled_tid:
                if self._polled_tid not in self._transactions:
                    # This connection is so old that we can no longer enumerate
                    # all the changes.
                    self._polled_tid = new_tid
                    return None

            changed_oids = set()
            for tid, txn in self._transactions.items(
                    self._polled_tid, new_tid,
                    excludemin=True, excludemax=False):
                if txn.status == 'p':
                    # This transaction has been packed, so it is no longer
                    # possible to enumerate all changed oids.
                    self._polled_tid = new_tid
                    return None
                if tid == self._ltid:
                    # ignore the transaction committed by this connection
                    continue
                changed_oids.update(txn.data.keys())

        self._polled_tid = self._ltid = new_tid
        return list(changed_oids)

    def tpc_finish(self, transaction, func=lambda tid: None):
        self._data_snapshot = None
        with self._main_lock:
            return MappingStorage.tpc_finish(self, transaction, func)

    def tpc_abort(self, transaction):
        self._data_snapshot = None
        with self._main_lock:
            MappingStorage.tpc_abort(self, transaction)

    def pack(self, t, referencesf, gc=True):
        # prevent all concurrent commits during packing
        with self._commit_lock:
            MappingStorage.pack(self, t, referencesf, gc)

1	##############################################################################
2	#
3	# Copyright (c) Zope Corporation and Contributors.
4	# All Rights Reserved.
5	#
6	# This software is subject to the provisions of the Zope Public License,
7	# Version 2.1 (ZPL). A copy of the ZPL should accompany this distribution.
8	# THIS SOFTWARE IS PROVIDED "AS IS" AND ANY AND ALL EXPRESS OR IMPLIED
9	# WARRANTIES ARE DISCLAIMED, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
10	# WARRANTIES OF TITLE, MERCHANTABILITY, AGAINST INFRINGEMENT, AND FITNESS
11	# FOR A PARTICULAR PURPOSE
12	#
13	##############################################################################
14	"""An extension of MappingStorage that depends on polling.
15
16	Each Connection has its own view of the database. Polling updates each
17	connection's view.
18	"""
19
20	from zope.interface import implementer	1✔
21
22	import ZODB.POSException	1✔
23	import ZODB.utils	1✔
24	from ZODB.interfaces import IMVCCStorage	1✔
25	from ZODB.MappingStorage import MappingStorage	1✔
26
27
28	@implementer(IMVCCStorage)	1✔
29	class MVCCMappingStorage(MappingStorage):	1✔
30
31	def __init__(self, name="MVCC Mapping Storage"):	1✔
32	MappingStorage.__init__(self, name=name)	1✔
33	# _polled_tid contains the transaction ID at the last poll.
34	self._polled_tid = b''	1✔
35	self._data_snapshot = None # {oid->(state, tid)}	1✔
36	self._main_lock = self._lock	1✔
37
38	def new_instance(self):	1✔
39	"""Returns a storage instance that is a view of the same data.
40	"""
41	inst = MVCCMappingStorage(name=self.__name__)	1✔
42	# All instances share the same OID data, transaction log, commit lock,
43	# and OID sequence.
44	inst._data = self._data	1✔
45	inst._transactions = self._transactions	1✔
46	inst._commit_lock = self._commit_lock	1✔
47	inst.new_oid = self.new_oid	1✔
48	inst.pack = self.pack	1✔
49	inst.loadBefore = self.loadBefore	1✔
50	inst._ltid = self._ltid	1✔
51	inst._main_lock = self._lock	1✔
52	return inst	1✔
53
54	@ZODB.utils.locked(MappingStorage.opened)	1✔
55	def sync(self, force=False):	1✔
56	self._data_snapshot = None	1✔
57
58	def release(self):	1✔
59	pass	1✔
60
61	@ZODB.utils.locked(MappingStorage.opened)	1✔
62	def load(self, oid, version=''):	1✔
63	assert not version, "Versions are not supported"	1✔
64	if self._data_snapshot is None:	1✔
65	self.poll_invalidations()	1✔
66	info = self._data_snapshot.get(oid)	1✔
67	if info:	1✔
68	return info	1✔
69	raise ZODB.POSException.POSKeyError(oid)	1✔
70
71	def poll_invalidations(self):	1✔
72	"""Poll the storage for changes by other connections.
73	"""
74	# prevent changes to _transactions and _data during analysis
75	with self._main_lock:	1✔
76	if self._transactions:	1✔
77	new_tid = self._transactions.maxKey()	1✔
78	else:
79	new_tid = ZODB.utils.z64	1✔
80
81	# Copy the current data into a snapshot. This is obviously
82	# very inefficient for large storages, but it's good for
83	# tests.
84	self._data_snapshot = {}	1✔
85	for oid, tid_data in self._data.items():	1✔
86	if tid_data:	1!
87	tid = tid_data.maxKey()	1✔
88	self._data_snapshot[oid] = tid_data[tid], tid	1✔
89
90	if self._polled_tid:	1✔
91	if self._polled_tid not in self._transactions:	1✔
92	# This connection is so old that we can no longer enumerate
93	# all the changes.
94	self._polled_tid = new_tid	1✔
95	return None	1✔
96
97	changed_oids = set()	1✔
98	for tid, txn in self._transactions.items(	1✔
99	self._polled_tid, new_tid,
100	excludemin=True, excludemax=False):
101	if txn.status == 'p':	1!
102	# This transaction has been packed, so it is no longer
103	# possible to enumerate all changed oids.
104	self._polled_tid = new_tid	×
105	return None	×
106	if tid == self._ltid:	1✔
107	# ignore the transaction committed by this connection
108	continue	1✔
109	changed_oids.update(txn.data.keys())	1✔
110
111	self._polled_tid = self._ltid = new_tid	1✔
112	return list(changed_oids)	1✔
113
114	def tpc_finish(self, transaction, func=lambda tid: None):	1✔
115	self._data_snapshot = None	1✔
116	with self._main_lock:	1✔
117	return MappingStorage.tpc_finish(self, transaction, func)	1✔
118
119	def tpc_abort(self, transaction):	1✔
120	self._data_snapshot = None	1✔
121	with self._main_lock:	1✔
122	MappingStorage.tpc_abort(self, transaction)	1✔
123
124	def pack(self, t, referencesf, gc=True):	1✔
125	# prevent all concurrent commits during packing
126	with self._commit_lock:	1✔
127	MappingStorage.pack(self, t, referencesf, gc)	1✔

zopefoundation / ZODB / 18443719934

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous