1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190
|
""" A very simple cell dict implementation using a version tag. The dictionary
maps keys to objects. If a specific key is changed a lot, a level of
indirection is introduced to make the version tag change less often.
"""
from rpython.rlib import jit, rerased, objectmodel
from pypy.interpreter.baseobjspace import W_Root
from pypy.interpreter.error import OperationError
from pypy.objspace.std.dictmultiobject import (
DictStrategy, ObjectDictStrategy, _never_equal_to_string,
create_iterator_classes)
from pypy.objspace.std.typeobject import (
MutableCell, IntMutableCell, ObjectMutableCell, write_cell)
class VersionTag(object):
pass
def unwrap_cell(space, w_value):
if isinstance(w_value, MutableCell):
return w_value.unwrap_cell(space)
return w_value
def _wrapkey(space, key):
return space.newtext(key)
class ModuleDictStrategy(DictStrategy):
erase, unerase = rerased.new_erasing_pair("modulecell")
erase = staticmethod(erase)
unerase = staticmethod(unerase)
_immutable_fields_ = ["version?"]
def __init__(self, space):
self.space = space
self.version = VersionTag()
def get_empty_storage(self):
return self.erase({})
def mutated(self):
self.version = VersionTag()
def getdictvalue_no_unwrapping(self, w_dict, key):
# NB: it's important to promote self here, so that self.version is a
# no-op due to the quasi-immutable field
self = jit.promote(self)
return self._getdictvalue_no_unwrapping_pure(self.version, w_dict, key)
@jit.elidable_promote('0,1,2')
def _getdictvalue_no_unwrapping_pure(self, version, w_dict, key):
return self.unerase(w_dict.dstorage).get(key, None)
def try_unwrap_key(self, space, w_key):
if space.is_w(space.type(w_key), space.w_text):
try:
return space.text_w(w_key)
except OperationError as e:
if e.match(space, space.w_UnicodeEncodeError):
return None
raise
return None
def setitem(self, w_dict, w_key, w_value):
space = self.space
key = self.try_unwrap_key(space, w_key)
if key is not None:
self.setitem_str(w_dict, key, w_value)
else:
self.switch_to_object_strategy(w_dict)
w_dict.setitem(w_key, w_value)
def setitem_str(self, w_dict, key, w_value):
cell = self.getdictvalue_no_unwrapping(w_dict, key)
return self._setitem_str_cell_known(cell, w_dict, key, w_value)
def _setitem_str_cell_known(self, cell, w_dict, key, w_value):
w_value = write_cell(self.space, cell, w_value)
if w_value is None:
return
self.mutated()
self.unerase(w_dict.dstorage)[key] = w_value
def setdefault(self, w_dict, w_key, w_default):
space = self.space
key = self.try_unwrap_key(space, w_key)
if key is not None:
cell = self.getdictvalue_no_unwrapping(w_dict, key)
w_result = unwrap_cell(self.space, cell)
if w_result is not None:
return w_result
self._setitem_str_cell_known(cell, w_dict, key, w_default)
return w_default
else:
self.switch_to_object_strategy(w_dict)
return w_dict.setdefault(w_key, w_default)
def delitem(self, w_dict, w_key):
space = self.space
w_key_type = space.type(w_key)
key = self.try_unwrap_key(space, w_key)
if key is not None:
dict_w = self.unerase(w_dict.dstorage)
try:
del dict_w[key]
except KeyError:
raise
else:
self.mutated()
elif _never_equal_to_string(space, w_key_type):
raise KeyError
else:
self.switch_to_object_strategy(w_dict)
w_dict.delitem(w_key)
def length(self, w_dict):
return len(self.unerase(w_dict.dstorage))
def getitem(self, w_dict, w_key):
space = self.space
w_lookup_type = space.type(w_key)
key = self.try_unwrap_key(space, w_key)
if key is not None:
return self.getitem_str(w_dict, key)
elif _never_equal_to_string(space, w_lookup_type):
return None
else:
self.switch_to_object_strategy(w_dict)
return w_dict.getitem(w_key)
def getitem_str(self, w_dict, key):
cell = self.getdictvalue_no_unwrapping(w_dict, key)
return unwrap_cell(self.space, cell)
def w_keys(self, w_dict):
space = self.space
l = self.unerase(w_dict.dstorage).keys()
return space.newlist_text(l)
def values(self, w_dict):
iterator = self.unerase(w_dict.dstorage).itervalues
return [unwrap_cell(self.space, cell) for cell in iterator()]
def items(self, w_dict):
space = self.space
iterator = self.unerase(w_dict.dstorage).iteritems
return [space.newtuple([_wrapkey(space, key), unwrap_cell(self.space, cell)])
for key, cell in iterator()]
def clear(self, w_dict):
self.unerase(w_dict.dstorage).clear()
self.mutated()
def popitem(self, w_dict):
space = self.space
d = self.unerase(w_dict.dstorage)
key, cell = d.popitem()
self.mutated()
return _wrapkey(space, key), unwrap_cell(self.space, cell)
def switch_to_object_strategy(self, w_dict):
space = self.space
d = self.unerase(w_dict.dstorage)
strategy = space.fromcache(ObjectDictStrategy)
d_new = strategy.unerase(strategy.get_empty_storage())
for key, cell in d.iteritems():
d_new[_wrapkey(space, key)] = unwrap_cell(self.space, cell)
w_dict.set_strategy(strategy)
w_dict.dstorage = strategy.erase(d_new)
def getiterkeys(self, w_dict):
return self.unerase(w_dict.dstorage).iterkeys()
def getitervalues(self, w_dict):
return self.unerase(w_dict.dstorage).itervalues()
def getiteritems_with_hash(self, w_dict):
return objectmodel.iteritems_with_hash(self.unerase(w_dict.dstorage))
wrapkey = _wrapkey
def wrapvalue(space, value):
return unwrap_cell(space, value)
create_iterator_classes(ModuleDictStrategy)
|