1
# Copyright (C) 2008, 2009 Canonical Ltd
3
# This program is free software; you can redistribute it and/or modify
4
# it under the terms of the GNU General Public License as published by
5
# the Free Software Foundation; either version 2 of the License, or
6
# (at your option) any later version.
8
# This program is distributed in the hope that it will be useful,
9
# but WITHOUT ANY WARRANTY; without even the implied warranty of
10
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
11
# GNU General Public License for more details.
13
# You should have received a copy of the GNU General Public License
14
# along with this program; if not, write to the Free Software
15
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
17
"""Persistent maps from tuple_of_strings->string using CHK stores.
19
Overview and current status:
21
The CHKMap class implements a dict from tuple_of_strings->string by using a trie
22
with internal nodes of 8-bit fan out; The key tuples are mapped to strings by
23
joining them by \x00, and \x00 padding shorter keys out to the length of the
24
longest key. Leaf nodes are packed as densely as possible, and internal nodes
25
are all an additional 8-bits wide leading to a sparse upper tree.
27
Updates to a CHKMap are done preferentially via the apply_delta method, to
28
allow optimisation of the update operation; but individual map/unmap calls are
29
possible and supported. Individual changes via map/unmap are buffered in memory
30
until the _save method is called to force serialisation of the tree.
31
apply_delta records its changes immediately by performing an implicit _save.
36
Densely packed upper nodes.
42
from bzrlib import lazy_import
43
lazy_import.lazy_import(globals(), """
55
from bzrlib.static_tuple import StaticTuple
58
# If each line is 50 bytes, and you have 255 internal pages, with 255-way fan
59
# out, it takes 3.1MB to cache the layer.
60
_PAGE_CACHE_SIZE = 4*1024*1024
61
# We are caching bytes so len(value) is perfectly accurate
62
_page_cache = lru_cache.LRUSizeCache(_PAGE_CACHE_SIZE)
67
# If a ChildNode falls below this many bytes, we check for a remap
68
_INTERESTING_NEW_SIZE = 50
69
# If a ChildNode shrinks by more than this amount, we check for a remap
70
_INTERESTING_SHRINKAGE_LIMIT = 20
71
# If we delete more than this many nodes applying a delta, we check for a remap
72
_INTERESTING_DELETES_LIMIT = 5
75
def _search_key_plain(key):
76
"""Map the key tuple into a search string that just uses the key bytes."""
77
return '\x00'.join(key)
80
search_key_registry = registry.Registry()
81
search_key_registry.register('plain', _search_key_plain)
85
"""A persistent map from string to string backed by a CHK store."""
87
__slots__ = ('_store', '_root_node', '_search_key_func')
89
def __init__(self, store, root_key, search_key_func=None):
90
"""Create a CHKMap object.
92
:param store: The store the CHKMap is stored in.
93
:param root_key: The root key of the map. None to create an empty
95
:param search_key_func: A function mapping a key => bytes. These bytes
96
are then used by the internal nodes to split up leaf nodes into
100
if search_key_func is None:
101
search_key_func = _search_key_plain
102
self._search_key_func = search_key_func
104
self._root_node = LeafNode(search_key_func=search_key_func)
106
self._root_node = self._node_key(root_key)
108
def apply_delta(self, delta):
109
"""Apply a delta to the map.
111
:param delta: An iterable of old_key, new_key, new_value tuples.
112
If new_key is not None, then new_key->new_value is inserted
113
into the map; if old_key is not None, then the old mapping
114
of old_key is removed.
117
# Check preconditions first.
118
as_st = StaticTuple.from_sequence
119
new_items = set([as_st(key) for (old, key, value) in delta
120
if key is not None and old is None])
121
existing_new = list(self.iteritems(key_filter=new_items))
123
raise errors.InconsistentDeltaDelta(delta,
124
"New items are already in the map %r." % existing_new)
126
for old, new, value in delta:
127
if old is not None and old != new:
128
self.unmap(old, check_remap=False)
130
for old, new, value in delta:
133
if delete_count > _INTERESTING_DELETES_LIMIT:
134
trace.mutter("checking remap as %d deletions", delete_count)
138
def _ensure_root(self):
139
"""Ensure that the root node is an object not a key."""
140
if type(self._root_node) is StaticTuple:
141
# Demand-load the root
142
self._root_node = self._get_node(self._root_node)
144
def _get_node(self, node):
147
Note that this does not update the _items dict in objects containing a
148
reference to this node. As such it does not prevent subsequent IO being
151
:param node: A tuple key or node object.
152
:return: A node object.
154
if type(node) is StaticTuple:
155
bytes = self._read_bytes(node)
156
return _deserialise(bytes, node,
157
search_key_func=self._search_key_func)
161
def _read_bytes(self, key):
163
return _page_cache[key]
165
stream = self._store.get_record_stream([key], 'unordered', True)
166
bytes = stream.next().get_bytes_as('fulltext')
167
_page_cache[key] = bytes
170
def _dump_tree(self, include_keys=False):
171
"""Return the tree in a string representation."""
173
res = self._dump_tree_node(self._root_node, prefix='', indent='',
174
include_keys=include_keys)
175
res.append('') # Give a trailing '\n'
176
return '\n'.join(res)
178
def _dump_tree_node(self, node, prefix, indent, include_keys=True):
179
"""For this node and all children, generate a string representation."""
184
node_key = node.key()
185
if node_key is not None:
186
key_str = ' %s' % (node_key[0],)
189
result.append('%s%r %s%s' % (indent, prefix, node.__class__.__name__,
191
if type(node) is InternalNode:
192
# Trigger all child nodes to get loaded
193
list(node._iter_nodes(self._store))
194
for prefix, sub in sorted(node._items.iteritems()):
195
result.extend(self._dump_tree_node(sub, prefix, indent + ' ',
196
include_keys=include_keys))
198
for key, value in sorted(node._items.iteritems()):
199
# Don't use prefix nor indent here to line up when used in
200
# tests in conjunction with assertEqualDiff
201
result.append(' %r %r' % (tuple(key), value))
205
def from_dict(klass, store, initial_value, maximum_size=0, key_width=1,
206
search_key_func=None):
207
"""Create a CHKMap in store with initial_value as the content.
209
:param store: The store to record initial_value in, a VersionedFiles
210
object with 1-tuple keys supporting CHK key generation.
211
:param initial_value: A dict to store in store. Its keys and values
213
:param maximum_size: The maximum_size rule to apply to nodes. This
214
determines the size at which no new data is added to a single node.
215
:param key_width: The number of elements in each key_tuple being stored
217
:param search_key_func: A function mapping a key => bytes. These bytes
218
are then used by the internal nodes to split up leaf nodes into
220
:return: The root chk of the resulting CHKMap.
222
root_key = klass._create_directly(store, initial_value,
223
maximum_size=maximum_size, key_width=key_width,
224
search_key_func=search_key_func)
225
assert type(root_key) is StaticTuple
229
def _create_via_map(klass, store, initial_value, maximum_size=0,
230
key_width=1, search_key_func=None):
231
result = klass(store, None, search_key_func=search_key_func)
232
result._root_node.set_maximum_size(maximum_size)
233
result._root_node._key_width = key_width
235
for key, value in initial_value.items():
236
delta.append((None, key, value))
237
root_key = result.apply_delta(delta)
241
def _create_directly(klass, store, initial_value, maximum_size=0,
242
key_width=1, search_key_func=None):
243
node = LeafNode(search_key_func=search_key_func)
244
node.set_maximum_size(maximum_size)
245
node._key_width = key_width
246
as_st = StaticTuple.from_sequence
247
node._items = dict([(as_st(key), val) for key, val
248
in initial_value.iteritems()])
249
node._raw_size = sum([node._key_value_len(key, value)
250
for key,value in node._items.iteritems()])
251
node._len = len(node._items)
252
node._compute_search_prefix()
253
node._compute_serialised_prefix()
256
and node._current_size() > maximum_size):
257
prefix, node_details = node._split(store)
258
if len(node_details) == 1:
259
raise AssertionError('Failed to split using node._split')
260
node = InternalNode(prefix, search_key_func=search_key_func)
261
node.set_maximum_size(maximum_size)
262
node._key_width = key_width
263
for split, subnode in node_details:
264
node.add_node(split, subnode)
265
keys = list(node.serialise(store))
267
assert (type(root_node) is StaticTuple
268
and len(root_node) == 1 and
269
type(root_node[0]) is str)
272
def iter_changes(self, basis):
273
"""Iterate over the changes between basis and self.
275
:return: An iterator of tuples: (key, old_value, new_value). Old_value
276
is None for keys only in self; new_value is None for keys only in
280
# Read both trees in lexographic, highest-first order.
281
# Any identical nodes we skip
282
# Any unique prefixes we output immediately.
283
# values in a leaf node are treated as single-value nodes in the tree
284
# which allows them to be not-special-cased. We know to output them
285
# because their value is a string, not a key(tuple) or node.
287
# corner cases to beware of when considering this function:
288
# *) common references are at different heights.
289
# consider two trees:
290
# {'a': LeafNode={'aaa':'foo', 'aab':'bar'}, 'b': LeafNode={'b'}}
291
# {'a': InternalNode={'aa':LeafNode={'aaa':'foo', 'aab':'bar'},
292
# 'ab':LeafNode={'ab':'bar'}}
293
# 'b': LeafNode={'b'}}
294
# the node with aaa/aab will only be encountered in the second tree
295
# after reading the 'a' subtree, but it is encountered in the first
296
# tree immediately. Variations on this may have read internal nodes
297
# like this. we want to cut the entire pending subtree when we
298
# realise we have a common node. For this we use a list of keys -
299
# the path to a node - and check the entire path is clean as we
301
if self._node_key(self._root_node) == self._node_key(basis._root_node):
305
excluded_keys = set()
306
self_node = self._root_node
307
basis_node = basis._root_node
308
# A heap, each element is prefix, node(tuple/NodeObject/string),
309
# key_path (a list of tuples, tail-sharing down the tree.)
312
def process_node(node, path, a_map, pending):
313
# take a node and expand it
314
node = a_map._get_node(node)
315
if type(node) == LeafNode:
316
path = (node._key, path)
317
for key, value in node._items.items():
318
# For a LeafNode, the key is a serialized_key, rather than
319
# a search_key, but the heap is using search_keys
320
search_key = node._search_key_func(key)
321
heapq.heappush(pending, (search_key, key, value, path))
323
# type(node) == InternalNode
324
path = (node._key, path)
325
for prefix, child in node._items.items():
326
heapq.heappush(pending, (prefix, None, child, path))
327
def process_common_internal_nodes(self_node, basis_node):
328
self_items = set(self_node._items.items())
329
basis_items = set(basis_node._items.items())
330
path = (self_node._key, None)
331
for prefix, child in self_items - basis_items:
332
heapq.heappush(self_pending, (prefix, None, child, path))
333
path = (basis_node._key, None)
334
for prefix, child in basis_items - self_items:
335
heapq.heappush(basis_pending, (prefix, None, child, path))
336
def process_common_leaf_nodes(self_node, basis_node):
337
self_items = set(self_node._items.items())
338
basis_items = set(basis_node._items.items())
339
path = (self_node._key, None)
340
for key, value in self_items - basis_items:
341
prefix = self._search_key_func(key)
342
heapq.heappush(self_pending, (prefix, key, value, path))
343
path = (basis_node._key, None)
344
for key, value in basis_items - self_items:
345
prefix = basis._search_key_func(key)
346
heapq.heappush(basis_pending, (prefix, key, value, path))
347
def process_common_prefix_nodes(self_node, self_path,
348
basis_node, basis_path):
349
# Would it be more efficient if we could request both at the same
351
self_node = self._get_node(self_node)
352
basis_node = basis._get_node(basis_node)
353
if (type(self_node) == InternalNode
354
and type(basis_node) == InternalNode):
355
# Matching internal nodes
356
process_common_internal_nodes(self_node, basis_node)
357
elif (type(self_node) == LeafNode
358
and type(basis_node) == LeafNode):
359
process_common_leaf_nodes(self_node, basis_node)
361
process_node(self_node, self_path, self, self_pending)
362
process_node(basis_node, basis_path, basis, basis_pending)
363
process_common_prefix_nodes(self_node, None, basis_node, None)
366
excluded_keys = set()
367
def check_excluded(key_path):
368
# Note that this is N^2, it depends on us trimming trees
369
# aggressively to not become slow.
370
# A better implementation would probably have a reverse map
371
# back to the children of a node, and jump straight to it when
372
# a common node is detected, the proceed to remove the already
373
# pending children. bzrlib.graph has a searcher module with a
375
while key_path is not None:
376
key, key_path = key_path
377
if key in excluded_keys:
382
while self_pending or basis_pending:
385
# self is exhausted: output remainder of basis
386
for prefix, key, node, path in basis_pending:
387
if check_excluded(path):
389
node = basis._get_node(node)
392
yield (key, node, None)
394
# subtree - fastpath the entire thing.
395
for key, value in node.iteritems(basis._store):
396
yield (key, value, None)
398
elif not basis_pending:
399
# basis is exhausted: output remainder of self.
400
for prefix, key, node, path in self_pending:
401
if check_excluded(path):
403
node = self._get_node(node)
406
yield (key, None, node)
408
# subtree - fastpath the entire thing.
409
for key, value in node.iteritems(self._store):
410
yield (key, None, value)
413
# XXX: future optimisation - yield the smaller items
414
# immediately rather than pushing everything on/off the
415
# heaps. Applies to both internal nodes and leafnodes.
416
if self_pending[0][0] < basis_pending[0][0]:
418
prefix, key, node, path = heapq.heappop(self_pending)
419
if check_excluded(path):
423
yield (key, None, node)
425
process_node(node, path, self, self_pending)
427
elif self_pending[0][0] > basis_pending[0][0]:
429
prefix, key, node, path = heapq.heappop(basis_pending)
430
if check_excluded(path):
434
yield (key, node, None)
436
process_node(node, path, basis, basis_pending)
439
# common prefix: possibly expand both
440
if self_pending[0][1] is None:
445
if basis_pending[0][1] is None:
450
if not read_self and not read_basis:
451
# compare a common value
452
self_details = heapq.heappop(self_pending)
453
basis_details = heapq.heappop(basis_pending)
454
if self_details[2] != basis_details[2]:
455
yield (self_details[1],
456
basis_details[2], self_details[2])
458
# At least one side wasn't a simple value
459
if (self._node_key(self_pending[0][2]) ==
460
self._node_key(basis_pending[0][2])):
461
# Identical pointers, skip (and don't bother adding to
462
# excluded, it won't turn up again.
463
heapq.heappop(self_pending)
464
heapq.heappop(basis_pending)
466
# Now we need to expand this node before we can continue
467
if read_self and read_basis:
468
# Both sides start with the same prefix, so process
470
self_prefix, _, self_node, self_path = heapq.heappop(
472
basis_prefix, _, basis_node, basis_path = heapq.heappop(
474
if self_prefix != basis_prefix:
475
raise AssertionError(
476
'%r != %r' % (self_prefix, basis_prefix))
477
process_common_prefix_nodes(
478
self_node, self_path,
479
basis_node, basis_path)
482
prefix, key, node, path = heapq.heappop(self_pending)
483
if check_excluded(path):
485
process_node(node, path, self, self_pending)
487
prefix, key, node, path = heapq.heappop(basis_pending)
488
if check_excluded(path):
490
process_node(node, path, basis, basis_pending)
493
def iteritems(self, key_filter=None):
494
"""Iterate over the entire CHKMap's contents."""
496
# TODO: StaticTuple Barrier here
497
if key_filter is not None:
498
as_st = StaticTuple.from_sequence
499
key_filter = [as_st(key) for key in key_filter]
500
return self._root_node.iteritems(self._store, key_filter=key_filter)
503
"""Return the key for this map."""
504
if type(self._root_node) is StaticTuple:
505
_check_key(self._root_node)
506
return self._root_node
508
return self._root_node._key
512
return len(self._root_node)
514
def map(self, key, value):
515
"""Map a key tuple to value.
517
:param key: A key to map.
518
:param value: The value to assign to key.
520
key = StaticTuple.from_sequence(key)
521
# Need a root object.
523
prefix, node_details = self._root_node.map(self._store, key, value)
524
if len(node_details) == 1:
525
self._root_node = node_details[0][1]
527
self._root_node = InternalNode(prefix,
528
search_key_func=self._search_key_func)
529
self._root_node.set_maximum_size(node_details[0][1].maximum_size)
530
self._root_node._key_width = node_details[0][1]._key_width
531
for split, node in node_details:
532
self._root_node.add_node(split, node)
534
def _node_key(self, node):
535
"""Get the key for a node whether it's a tuple or node."""
536
if type(node) is tuple:
537
node = StaticTuple.from_sequence(node)
538
if type(node) is StaticTuple:
544
def unmap(self, key, check_remap=True):
545
"""remove key from the map."""
546
key = StaticTuple.from_sequence(key)
548
if type(self._root_node) is InternalNode:
549
unmapped = self._root_node.unmap(self._store, key,
550
check_remap=check_remap)
552
unmapped = self._root_node.unmap(self._store, key)
553
self._root_node = unmapped
555
def _check_remap(self):
556
"""Check if nodes can be collapsed."""
558
if type(self._root_node) is InternalNode:
559
self._root_node._check_remap(self._store)
562
"""Save the map completely.
564
:return: The key of the root node.
566
if type(self._root_node) is StaticTuple:
568
return self._root_node
569
keys = list(self._root_node.serialise(self._store))
570
assert type(keys[-1]) is StaticTuple
575
"""Base class defining the protocol for CHK Map nodes.
577
:ivar _raw_size: The total size of the serialized key:value data, before
578
adding the header bytes, and without prefix compression.
581
__slots__ = ('_key', '_len', '_maximum_size', '_key_width',
582
'_raw_size', '_items', '_search_prefix', '_search_key_func'
585
def __init__(self, key_width=1):
588
:param key_width: The width of keys for this node.
591
# Current number of elements
593
self._maximum_size = 0
594
self._key_width = key_width
595
# current size in bytes
597
# The pointers/values this node has - meaning defined by child classes.
599
# The common search prefix
600
self._search_prefix = None
603
items_str = str(sorted(self._items))
604
if len(items_str) > 20:
605
items_str = items_str[:16] + '...]'
606
return '%s(key:%s len:%s size:%s max:%s prefix:%s items:%s)' % (
607
self.__class__.__name__, self._key, self._len, self._raw_size,
608
self._maximum_size, self._search_prefix, items_str)
617
def maximum_size(self):
618
"""What is the upper limit for adding references to a node."""
619
return self._maximum_size
621
def set_maximum_size(self, new_size):
622
"""Set the size threshold for nodes.
624
:param new_size: The size at which no data is added to a node. 0 for
627
self._maximum_size = new_size
630
def common_prefix(cls, prefix, key):
631
"""Given 2 strings, return the longest prefix common to both.
633
:param prefix: This has been the common prefix for other keys, so it is
634
more likely to be the common prefix in this case as well.
635
:param key: Another string to compare to
637
if key.startswith(prefix):
640
# Is there a better way to do this?
641
for pos, (left, right) in enumerate(zip(prefix, key)):
645
common = prefix[:pos+1]
649
def common_prefix_for_keys(cls, keys):
650
"""Given a list of keys, find their common prefix.
652
:param keys: An iterable of strings.
653
:return: The longest common prefix of all keys.
657
if common_prefix is None:
660
common_prefix = cls.common_prefix(common_prefix, key)
661
if not common_prefix:
662
# if common_prefix is the empty string, then we know it won't
668
# Singleton indicating we have not computed _search_prefix yet
671
class LeafNode(Node):
672
"""A node containing actual key:value pairs.
674
:ivar _items: A dict of key->value items. The key is in tuple form.
675
:ivar _size: The number of bytes that would be used by serializing all of
679
__slots__ = ('_common_serialised_prefix', '_serialise_key')
681
def __init__(self, search_key_func=None):
683
# All of the keys in this leaf node share this common prefix
684
self._common_serialised_prefix = None
685
self._serialise_key = '\x00'.join
686
if search_key_func is None:
687
self._search_key_func = _search_key_plain
689
self._search_key_func = search_key_func
692
items_str = str(sorted(self._items))
693
if len(items_str) > 20:
694
items_str = items_str[:16] + '...]'
696
'%s(key:%s len:%s size:%s max:%s prefix:%s keywidth:%s items:%s)' \
697
% (self.__class__.__name__, self._key, self._len, self._raw_size,
698
self._maximum_size, self._search_prefix, self._key_width, items_str)
700
def _current_size(self):
701
"""Answer the current serialised size of this node.
703
This differs from self._raw_size in that it includes the bytes used for
706
if self._common_serialised_prefix is None:
710
# We will store a single string with the common prefix
711
# And then that common prefix will not be stored in any of the
713
prefix_len = len(self._common_serialised_prefix)
714
bytes_for_items = (self._raw_size - (prefix_len * self._len))
715
return (9 # 'chkleaf:\n'
716
+ len(str(self._maximum_size)) + 1
717
+ len(str(self._key_width)) + 1
718
+ len(str(self._len)) + 1
723
def deserialise(klass, bytes, key, search_key_func=None):
724
"""Deserialise bytes, with key key, into a LeafNode.
726
:param bytes: The bytes of the node.
727
:param key: The key that the serialised node has.
729
return _deserialise_leaf_node(bytes, key,
730
search_key_func=search_key_func)
732
def iteritems(self, store, key_filter=None):
733
"""Iterate over items in the node.
735
:param key_filter: A filter to apply to the node. It should be a
736
list/set/dict or similar repeatedly iterable container.
738
if key_filter is not None:
739
# Adjust the filter - short elements go to a prefix filter. All
740
# other items are looked up directly.
741
# XXX: perhaps defaultdict? Profiling<rinse and repeat>
743
for key in key_filter:
744
if len(key) == self._key_width:
745
# This filter is meant to match exactly one key, yield it
748
yield key, self._items[key]
750
# This key is not present in this map, continue
753
# Short items, we need to match based on a prefix
754
length_filter = filters.setdefault(len(key), set())
755
length_filter.add(key)
757
filters = filters.items()
758
for item in self._items.iteritems():
759
for length, length_filter in filters:
760
if item[0][:length] in length_filter:
764
for item in self._items.iteritems():
767
def _key_value_len(self, key, value):
768
# TODO: Should probably be done without actually joining the key, but
769
# then that can be done via the C extension
770
return (len(self._serialise_key(key)) + 1
771
+ len(str(value.count('\n'))) + 1
774
def _search_key(self, key):
775
return self._search_key_func(key)
777
def _map_no_split(self, key, value):
778
"""Map a key to a value.
780
This assumes either the key does not already exist, or you have already
781
removed its size and length from self.
783
:return: True if adding this node should cause us to split.
785
self._items[key] = value
786
self._raw_size += self._key_value_len(key, value)
788
serialised_key = self._serialise_key(key)
789
if self._common_serialised_prefix is None:
790
self._common_serialised_prefix = serialised_key
792
self._common_serialised_prefix = self.common_prefix(
793
self._common_serialised_prefix, serialised_key)
794
search_key = self._search_key(key)
795
if self._search_prefix is _unknown:
796
self._compute_search_prefix()
797
if self._search_prefix is None:
798
self._search_prefix = search_key
800
self._search_prefix = self.common_prefix(
801
self._search_prefix, search_key)
803
and self._maximum_size
804
and self._current_size() > self._maximum_size):
805
# Check to see if all of the search_keys for this node are
806
# identical. We allow the node to grow under that circumstance
807
# (we could track this as common state, but it is infrequent)
808
if (search_key != self._search_prefix
809
or not self._are_search_keys_identical()):
813
def _split(self, store):
814
"""We have overflowed.
816
Split this node into multiple LeafNodes, return it up the stack so that
817
the next layer creates a new InternalNode and references the new nodes.
819
:return: (common_serialised_prefix, [(node_serialised_prefix, node)])
821
if self._search_prefix is _unknown:
822
raise AssertionError('Search prefix must be known')
823
common_prefix = self._search_prefix
824
split_at = len(common_prefix) + 1
826
for key, value in self._items.iteritems():
827
search_key = self._search_key(key)
828
prefix = search_key[:split_at]
829
# TODO: Generally only 1 key can be exactly the right length,
830
# which means we can only have 1 key in the node pointed
831
# at by the 'prefix\0' key. We might want to consider
832
# folding it into the containing InternalNode rather than
833
# having a fixed length-1 node.
834
# Note this is probably not true for hash keys, as they
835
# may get a '\00' node anywhere, but won't have keys of
837
if len(prefix) < split_at:
838
prefix += '\x00'*(split_at - len(prefix))
839
if prefix not in result:
840
node = LeafNode(search_key_func=self._search_key_func)
841
node.set_maximum_size(self._maximum_size)
842
node._key_width = self._key_width
843
result[prefix] = node
845
node = result[prefix]
846
sub_prefix, node_details = node.map(store, key, value)
847
if len(node_details) > 1:
848
if prefix != sub_prefix:
849
# This node has been split and is now found via a different
852
new_node = InternalNode(sub_prefix,
853
search_key_func=self._search_key_func)
854
new_node.set_maximum_size(self._maximum_size)
855
new_node._key_width = self._key_width
856
for split, node in node_details:
857
new_node.add_node(split, node)
858
result[prefix] = new_node
859
return common_prefix, result.items()
861
def map(self, store, key, value):
862
"""Map key to value."""
863
if key in self._items:
864
self._raw_size -= self._key_value_len(key, self._items[key])
867
if self._map_no_split(key, value):
868
return self._split(store)
870
if self._search_prefix is _unknown:
871
raise AssertionError('%r must be known' % self._search_prefix)
872
return self._search_prefix, [("", self)]
874
def serialise(self, store):
875
"""Serialise the LeafNode to store.
877
:param store: A VersionedFiles honouring the CHK extensions.
878
:return: An iterable of the keys inserted by this operation.
880
lines = ["chkleaf:\n"]
881
lines.append("%d\n" % self._maximum_size)
882
lines.append("%d\n" % self._key_width)
883
lines.append("%d\n" % self._len)
884
if self._common_serialised_prefix is None:
886
if len(self._items) != 0:
887
raise AssertionError('If _common_serialised_prefix is None'
888
' we should have no items')
890
lines.append('%s\n' % (self._common_serialised_prefix,))
891
prefix_len = len(self._common_serialised_prefix)
892
for key, value in sorted(self._items.items()):
893
# Always add a final newline
894
value_lines = osutils.chunks_to_lines([value + '\n'])
895
serialized = "%s\x00%s\n" % (self._serialise_key(key),
897
if not serialized.startswith(self._common_serialised_prefix):
898
raise AssertionError('We thought the common prefix was %r'
899
' but entry %r does not have it in common'
900
% (self._common_serialised_prefix, serialized))
901
lines.append(serialized[prefix_len:])
902
lines.extend(value_lines)
903
sha1, _, _ = store.add_lines((None,), (), lines)
904
self._key = StaticTuple("sha1:" + sha1,).intern()
905
bytes = ''.join(lines)
906
if len(bytes) != self._current_size():
907
raise AssertionError('Invalid _current_size')
908
_page_cache.add(self._key, bytes)
912
"""Return the references to other CHK's held by this node."""
915
def _compute_search_prefix(self):
916
"""Determine the common search prefix for all keys in this node.
918
:return: A bytestring of the longest search key prefix that is
919
unique within this node.
921
search_keys = [self._search_key_func(key) for key in self._items]
922
self._search_prefix = self.common_prefix_for_keys(search_keys)
923
return self._search_prefix
925
def _are_search_keys_identical(self):
926
"""Check to see if the search keys for all entries are the same.
928
When using a hash as the search_key it is possible for non-identical
929
keys to collide. If that happens enough, we may try overflow a
930
LeafNode, but as all are collisions, we must not split.
932
common_search_key = None
933
for key in self._items:
934
search_key = self._search_key(key)
935
if common_search_key is None:
936
common_search_key = search_key
937
elif search_key != common_search_key:
941
def _compute_serialised_prefix(self):
942
"""Determine the common prefix for serialised keys in this node.
944
:return: A bytestring of the longest serialised key prefix that is
945
unique within this node.
947
serialised_keys = [self._serialise_key(key) for key in self._items]
948
self._common_serialised_prefix = self.common_prefix_for_keys(
950
return self._common_serialised_prefix
952
def unmap(self, store, key):
953
"""Unmap key from the node."""
955
self._raw_size -= self._key_value_len(key, self._items[key])
957
trace.mutter("key %s not found in %r", key, self._items)
962
# Recompute from scratch
963
self._compute_search_prefix()
964
self._compute_serialised_prefix()
968
class InternalNode(Node):
969
"""A node that contains references to other nodes.
971
An InternalNode is responsible for mapping search key prefixes to child
974
:ivar _items: serialised_key => node dictionary. node may be a tuple,
975
LeafNode or InternalNode.
978
__slots__ = ('_node_width',)
980
def __init__(self, prefix='', search_key_func=None):
982
# The size of an internalnode with default values and no children.
983
# How many octets key prefixes within this node are.
985
self._search_prefix = prefix
986
if search_key_func is None:
987
self._search_key_func = _search_key_plain
989
self._search_key_func = search_key_func
991
def add_node(self, prefix, node):
992
"""Add a child node with prefix prefix, and node node.
994
:param prefix: The search key prefix for node.
995
:param node: The node being added.
997
if self._search_prefix is None:
998
raise AssertionError("_search_prefix should not be None")
999
if not prefix.startswith(self._search_prefix):
1000
raise AssertionError("prefixes mismatch: %s must start with %s"
1001
% (prefix,self._search_prefix))
1002
if len(prefix) != len(self._search_prefix) + 1:
1003
raise AssertionError("prefix wrong length: len(%s) is not %d" %
1004
(prefix, len(self._search_prefix) + 1))
1005
self._len += len(node)
1006
if not len(self._items):
1007
self._node_width = len(prefix)
1008
if self._node_width != len(self._search_prefix) + 1:
1009
raise AssertionError("node width mismatch: %d is not %d" %
1010
(self._node_width, len(self._search_prefix) + 1))
1011
self._items[prefix] = node
1014
def _current_size(self):
1015
"""Answer the current serialised size of this node."""
1016
return (self._raw_size + len(str(self._len)) + len(str(self._key_width)) +
1017
len(str(self._maximum_size)))
1020
def deserialise(klass, bytes, key, search_key_func=None):
1021
"""Deserialise bytes to an InternalNode, with key key.
1023
:param bytes: The bytes of the node.
1024
:param key: The key that the serialised node has.
1025
:return: An InternalNode instance.
1027
if type(key) is not StaticTuple:
1028
import pdb; pdb.set_trace()
1029
key = StaticTuple.from_sequence(key).intern()
1030
return _deserialise_internal_node(bytes, key,
1031
search_key_func=search_key_func)
1033
def iteritems(self, store, key_filter=None):
1034
for node, node_filter in self._iter_nodes(store, key_filter=key_filter):
1035
for item in node.iteritems(store, key_filter=node_filter):
1038
def _iter_nodes(self, store, key_filter=None, batch_size=None):
1039
"""Iterate over node objects which match key_filter.
1041
:param store: A store to use for accessing content.
1042
:param key_filter: A key filter to filter nodes. Only nodes that might
1043
contain a key in key_filter will be returned.
1044
:param batch_size: If not None, then we will return the nodes that had
1045
to be read using get_record_stream in batches, rather than reading
1047
:return: An iterable of nodes. This function does not have to be fully
1048
consumed. (There will be no pending I/O when items are being returned.)
1050
# Map from chk key ('sha1:...',) to (prefix, key_filter)
1051
# prefix is the key in self._items to use, key_filter is the key_filter
1052
# entries that would match this node
1055
if key_filter is None:
1056
# yielding all nodes, yield whatever we have, and queue up a read
1057
# for whatever we are missing
1059
for prefix, node in self._items.iteritems():
1060
if node.__class__ is StaticTuple:
1061
keys[node] = (prefix, None)
1064
elif len(key_filter) == 1:
1065
# Technically, this path could also be handled by the first check
1066
# in 'self._node_width' in length_filters. However, we can handle
1067
# this case without spending any time building up the
1068
# prefix_to_keys, etc state.
1070
# This is a bit ugly, but TIMEIT showed it to be by far the fastest
1071
# 0.626us list(key_filter)[0]
1072
# is a func() for list(), 2 mallocs, and a getitem
1073
# 0.489us [k for k in key_filter][0]
1074
# still has the mallocs, avoids the func() call
1075
# 0.350us iter(key_filter).next()
1076
# has a func() call, and mallocs an iterator
1077
# 0.125us for key in key_filter: pass
1078
# no func() overhead, might malloc an iterator
1079
# 0.105us for key in key_filter: break
1080
# no func() overhead, might malloc an iterator, probably
1081
# avoids checking an 'else' clause as part of the for
1082
for key in key_filter:
1084
search_prefix = self._search_prefix_filter(key)
1085
if len(search_prefix) == self._node_width:
1086
# This item will match exactly, so just do a dict lookup, and
1087
# see what we can return
1090
node = self._items[search_prefix]
1092
# A given key can only match 1 child node, if it isn't
1093
# there, then we can just return nothing
1095
if node.__class__ is StaticTuple:
1096
keys[node] = (search_prefix, [key])
1098
# This is loaded, and the only thing that can match,
1103
# First, convert all keys into a list of search prefixes
1104
# Aggregate common prefixes, and track the keys they come from
1107
for key in key_filter:
1108
search_prefix = self._search_prefix_filter(key)
1109
length_filter = length_filters.setdefault(
1110
len(search_prefix), set())
1111
length_filter.add(search_prefix)
1112
prefix_to_keys.setdefault(search_prefix, []).append(key)
1114
if (self._node_width in length_filters
1115
and len(length_filters) == 1):
1116
# all of the search prefixes match exactly _node_width. This
1117
# means that everything is an exact match, and we can do a
1118
# lookup into self._items, rather than iterating over the items
1120
search_prefixes = length_filters[self._node_width]
1121
for search_prefix in search_prefixes:
1123
node = self._items[search_prefix]
1125
# We can ignore this one
1127
node_key_filter = prefix_to_keys[search_prefix]
1128
if node.__class__ is StaticTuple:
1129
keys[node] = (search_prefix, node_key_filter)
1131
yield node, node_key_filter
1133
# The slow way. We walk every item in self._items, and check to
1134
# see if there are any matches
1135
length_filters = length_filters.items()
1136
for prefix, node in self._items.iteritems():
1137
node_key_filter = []
1138
for length, length_filter in length_filters:
1139
sub_prefix = prefix[:length]
1140
if sub_prefix in length_filter:
1141
node_key_filter.extend(prefix_to_keys[sub_prefix])
1142
if node_key_filter: # this key matched something, yield it
1143
if node.__class__ is StaticTuple:
1144
keys[node] = (prefix, node_key_filter)
1146
yield node, node_key_filter
1148
# Look in the page cache for some more bytes
1152
bytes = _page_cache[key]
1156
node = _deserialise(bytes, key,
1157
search_key_func=self._search_key_func)
1158
prefix, node_key_filter = keys[key]
1159
self._items[prefix] = node
1161
yield node, node_key_filter
1162
for key in found_keys:
1165
# demand load some pages.
1166
if batch_size is None:
1167
# Read all the keys in
1168
batch_size = len(keys)
1169
key_order = list(keys)
1170
for batch_start in range(0, len(key_order), batch_size):
1171
batch = key_order[batch_start:batch_start + batch_size]
1172
# We have to fully consume the stream so there is no pending
1173
# I/O, so we buffer the nodes for now.
1174
stream = store.get_record_stream(batch, 'unordered', True)
1175
node_and_filters = []
1176
for record in stream:
1177
bytes = record.get_bytes_as('fulltext')
1178
node = _deserialise(bytes, record.key,
1179
search_key_func=self._search_key_func)
1180
prefix, node_key_filter = keys[record.key]
1181
node_and_filters.append((node, node_key_filter))
1182
self._items[prefix] = node
1183
_page_cache.add(record.key, bytes)
1184
for info in node_and_filters:
1187
def map(self, store, key, value):
1188
"""Map key to value."""
1189
if not len(self._items):
1190
raise AssertionError("can't map in an empty InternalNode.")
1191
search_key = self._search_key(key)
1192
if self._node_width != len(self._search_prefix) + 1:
1193
raise AssertionError("node width mismatch: %d is not %d" %
1194
(self._node_width, len(self._search_prefix) + 1))
1195
if not search_key.startswith(self._search_prefix):
1196
# This key doesn't fit in this index, so we need to split at the
1197
# point where it would fit, insert self into that internal node,
1198
# and then map this key into that node.
1199
new_prefix = self.common_prefix(self._search_prefix,
1201
new_parent = InternalNode(new_prefix,
1202
search_key_func=self._search_key_func)
1203
new_parent.set_maximum_size(self._maximum_size)
1204
new_parent._key_width = self._key_width
1205
new_parent.add_node(self._search_prefix[:len(new_prefix)+1],
1207
return new_parent.map(store, key, value)
1208
children = [node for node, _
1209
in self._iter_nodes(store, key_filter=[key])]
1214
child = self._new_child(search_key, LeafNode)
1215
old_len = len(child)
1216
if type(child) is LeafNode:
1217
old_size = child._current_size()
1220
prefix, node_details = child.map(store, key, value)
1221
if len(node_details) == 1:
1222
# child may have shrunk, or might be a new node
1223
child = node_details[0][1]
1224
self._len = self._len - old_len + len(child)
1225
self._items[search_key] = child
1228
if type(child) is LeafNode:
1229
if old_size is None:
1230
# The old node was an InternalNode which means it has now
1231
# collapsed, so we need to check if it will chain to a
1232
# collapse at this level.
1233
trace.mutter("checking remap as InternalNode -> LeafNode")
1234
new_node = self._check_remap(store)
1236
# If the LeafNode has shrunk in size, we may want to run
1237
# a remap check. Checking for a remap is expensive though
1238
# and the frequency of a successful remap is very low.
1239
# Shrinkage by small amounts is common, so we only do the
1240
# remap check if the new_size is low or the shrinkage
1241
# amount is over a configurable limit.
1242
new_size = child._current_size()
1243
shrinkage = old_size - new_size
1244
if (shrinkage > 0 and new_size < _INTERESTING_NEW_SIZE
1245
or shrinkage > _INTERESTING_SHRINKAGE_LIMIT):
1247
"checking remap as size shrunk by %d to be %d",
1248
shrinkage, new_size)
1249
new_node = self._check_remap(store)
1250
if new_node._search_prefix is None:
1251
raise AssertionError("_search_prefix should not be None")
1252
return new_node._search_prefix, [('', new_node)]
1253
# child has overflown - create a new intermediate node.
1254
# XXX: This is where we might want to try and expand our depth
1255
# to refer to more bytes of every child (which would give us
1256
# multiple pointers to child nodes, but less intermediate nodes)
1257
child = self._new_child(search_key, InternalNode)
1258
child._search_prefix = prefix
1259
for split, node in node_details:
1260
child.add_node(split, node)
1261
self._len = self._len - old_len + len(child)
1263
return self._search_prefix, [("", self)]
1265
def _new_child(self, search_key, klass):
1266
"""Create a new child node of type klass."""
1268
child.set_maximum_size(self._maximum_size)
1269
child._key_width = self._key_width
1270
child._search_key_func = self._search_key_func
1271
self._items[search_key] = child
1274
def serialise(self, store):
1275
"""Serialise the node to store.
1277
:param store: A VersionedFiles honouring the CHK extensions.
1278
:return: An iterable of the keys inserted by this operation.
1280
for node in self._items.itervalues():
1281
if type(node) is StaticTuple:
1282
# Never deserialised.
1284
if node._key is not None:
1287
for key in node.serialise(store):
1289
lines = ["chknode:\n"]
1290
lines.append("%d\n" % self._maximum_size)
1291
lines.append("%d\n" % self._key_width)
1292
lines.append("%d\n" % self._len)
1293
if self._search_prefix is None:
1294
raise AssertionError("_search_prefix should not be None")
1295
lines.append('%s\n' % (self._search_prefix,))
1296
prefix_len = len(self._search_prefix)
1297
for prefix, node in sorted(self._items.items()):
1298
if type(node) is StaticTuple:
1302
serialised = "%s\x00%s\n" % (prefix, key)
1303
if not serialised.startswith(self._search_prefix):
1304
raise AssertionError("prefixes mismatch: %s must start with %s"
1305
% (serialised, self._search_prefix))
1306
lines.append(serialised[prefix_len:])
1307
sha1, _, _ = store.add_lines((None,), (), lines)
1308
self._key = StaticTuple("sha1:" + sha1,).intern()
1309
_page_cache.add(self._key, ''.join(lines))
1312
def _search_key(self, key):
1313
"""Return the serialised key for key in this node."""
1314
# search keys are fixed width. All will be self._node_width wide, so we
1316
return (self._search_key_func(key) + '\x00'*self._node_width)[:self._node_width]
1318
def _search_prefix_filter(self, key):
1319
"""Serialise key for use as a prefix filter in iteritems."""
1320
return self._search_key_func(key)[:self._node_width]
1322
def _split(self, offset):
1323
"""Split this node into smaller nodes starting at offset.
1325
:param offset: The offset to start the new child nodes at.
1326
:return: An iterable of (prefix, node) tuples. prefix is a byte
1327
prefix for reaching node.
1329
if offset >= self._node_width:
1330
for node in self._items.values():
1331
for result in node._split(offset):
1334
for key, node in self._items.items():
1338
"""Return the references to other CHK's held by this node."""
1339
if self._key is None:
1340
raise AssertionError("unserialised nodes have no refs.")
1342
for value in self._items.itervalues():
1343
if type(value) is StaticTuple:
1346
refs.append(value.key())
1349
def _compute_search_prefix(self, extra_key=None):
1350
"""Return the unique key prefix for this node.
1352
:return: A bytestring of the longest search key prefix that is
1353
unique within this node.
1355
self._search_prefix = self.common_prefix_for_keys(self._items)
1356
return self._search_prefix
1358
def unmap(self, store, key, check_remap=True):
1359
"""Remove key from this node and it's children."""
1360
if not len(self._items):
1361
raise AssertionError("can't unmap in an empty InternalNode.")
1362
children = [node for node, _
1363
in self._iter_nodes(store, key_filter=[key])]
1369
unmapped = child.unmap(store, key)
1371
search_key = self._search_key(key)
1372
if len(unmapped) == 0:
1373
# All child nodes are gone, remove the child:
1374
del self._items[search_key]
1377
# Stash the returned node
1378
self._items[search_key] = unmapped
1379
if len(self._items) == 1:
1380
# this node is no longer needed:
1381
return self._items.values()[0]
1382
if type(unmapped) is InternalNode:
1385
return self._check_remap(store)
1389
def _check_remap(self, store):
1390
"""Check if all keys contained by children fit in a single LeafNode.
1392
:param store: A store to use for reading more nodes
1393
:return: Either self, or a new LeafNode which should replace self.
1395
# Logic for how we determine when we need to rebuild
1396
# 1) Implicitly unmap() is removing a key which means that the child
1397
# nodes are going to be shrinking by some extent.
1398
# 2) If all children are LeafNodes, it is possible that they could be
1399
# combined into a single LeafNode, which can then completely replace
1400
# this internal node with a single LeafNode
1401
# 3) If *one* child is an InternalNode, we assume it has already done
1402
# all the work to determine that its children cannot collapse, and
1403
# we can then assume that those nodes *plus* the current nodes don't
1404
# have a chance of collapsing either.
1405
# So a very cheap check is to just say if 'unmapped' is an
1406
# InternalNode, we don't have to check further.
1408
# TODO: Another alternative is to check the total size of all known
1409
# LeafNodes. If there is some formula we can use to determine the
1410
# final size without actually having to read in any more
1411
# children, it would be nice to have. However, we have to be
1412
# careful with stuff like nodes that pull out the common prefix
1413
# of each key, as adding a new key can change the common prefix
1414
# and cause size changes greater than the length of one key.
1415
# So for now, we just add everything to a new Leaf until it
1416
# splits, as we know that will give the right answer
1417
new_leaf = LeafNode(search_key_func=self._search_key_func)
1418
new_leaf.set_maximum_size(self._maximum_size)
1419
new_leaf._key_width = self._key_width
1420
# A batch_size of 16 was chosen because:
1421
# a) In testing, a 4k page held 14 times. So if we have more than 16
1422
# leaf nodes we are unlikely to hold them in a single new leaf
1423
# node. This still allows for 1 round trip
1424
# b) With 16-way fan out, we can still do a single round trip
1425
# c) With 255-way fan out, we don't want to read all 255 and destroy
1426
# the page cache, just to determine that we really don't need it.
1427
for node, _ in self._iter_nodes(store, batch_size=16):
1428
if type(node) is InternalNode:
1429
# Without looking at any leaf nodes, we are sure
1431
for key, value in node._items.iteritems():
1432
if new_leaf._map_no_split(key, value):
1434
trace.mutter("remap generated a new LeafNode")
1438
def _deserialise(bytes, key, search_key_func):
1439
"""Helper for repositorydetails - convert bytes to a node."""
1440
if bytes.startswith("chkleaf:\n"):
1441
node = LeafNode.deserialise(bytes, key, search_key_func=search_key_func)
1442
elif bytes.startswith("chknode:\n"):
1443
node = InternalNode.deserialise(bytes, key,
1444
search_key_func=search_key_func)
1446
raise AssertionError("Unknown node type.")
1450
class CHKMapDifference(object):
1451
"""Iterate the stored pages and key,value pairs for (new - old).
1453
This class provides a generator over the stored CHK pages and the
1454
(key, value) pairs that are in any of the new maps and not in any of the
1457
Note that it may yield chk pages that are common (especially root nodes),
1458
but it won't yield (key,value) pairs that are common.
1461
def __init__(self, store, new_root_keys, old_root_keys,
1462
search_key_func, pb=None):
1464
self._new_root_keys = new_root_keys
1465
self._old_root_keys = old_root_keys
1467
# All uninteresting chks that we have seen. By the time they are added
1468
# here, they should be either fully ignored, or queued up for
1470
self._all_old_chks = set(self._old_root_keys)
1471
# All items that we have seen from the old_root_keys
1472
self._all_old_items = set()
1473
# These are interesting items which were either read, or already in the
1474
# interesting queue (so we don't need to walk them again)
1475
self._processed_new_refs = set()
1476
self._search_key_func = search_key_func
1478
# The uninteresting and interesting nodes to be searched
1479
self._old_queue = []
1480
self._new_queue = []
1481
# Holds the (key, value) items found when processing the root nodes,
1482
# waiting for the uninteresting nodes to be walked
1483
self._new_item_queue = []
1486
def _read_nodes_from_store(self, keys):
1487
# We chose not to use _page_cache, because we think in terms of records
1488
# to be yielded. Also, we expect to touch each page only 1 time during
1489
# this code. (We may want to evaluate saving the raw bytes into the
1490
# page cache, which would allow a working tree update after the fetch
1491
# to not have to read the bytes again.)
1492
stream = self._store.get_record_stream(keys, 'unordered', True)
1493
for record in stream:
1494
if self._pb is not None:
1496
if record.storage_kind == 'absent':
1497
raise errors.NoSuchRevision(self._store, record.key)
1498
bytes = record.get_bytes_as('fulltext')
1499
node = _deserialise(bytes, record.key,
1500
search_key_func=self._search_key_func)
1501
if type(node) is InternalNode:
1502
# Note we don't have to do node.refs() because we know that
1503
# there are no children that have been pushed into this node
1504
prefix_refs = node._items.items()
1508
items = node._items.items()
1509
yield record, node, prefix_refs, items
1511
def _read_old_roots(self):
1512
old_chks_to_enqueue = []
1513
all_old_chks = self._all_old_chks
1514
for record, node, prefix_refs, items in \
1515
self._read_nodes_from_store(self._old_root_keys):
1516
# Uninteresting node
1517
prefix_refs = [p_r for p_r in prefix_refs
1518
if p_r[1] not in all_old_chks]
1519
new_refs = [p_r[1] for p_r in prefix_refs]
1520
all_old_chks.update(new_refs)
1521
self._all_old_items.update(items)
1522
# Queue up the uninteresting references
1523
# Don't actually put them in the 'to-read' queue until we have
1524
# finished checking the interesting references
1525
old_chks_to_enqueue.extend(prefix_refs)
1526
return old_chks_to_enqueue
1528
def _enqueue_old(self, new_prefixes, old_chks_to_enqueue):
1529
# At this point, we have read all the uninteresting and interesting
1530
# items, so we can queue up the uninteresting stuff, knowing that we've
1531
# handled the interesting ones
1532
for prefix, ref in old_chks_to_enqueue:
1533
not_interesting = True
1534
for i in xrange(len(prefix), 0, -1):
1535
if prefix[:i] in new_prefixes:
1536
not_interesting = False
1539
# This prefix is not part of the remaining 'interesting set'
1541
self._old_queue.append(ref)
1543
def _read_all_roots(self):
1544
"""Read the root pages.
1546
This is structured as a generator, so that the root records can be
1547
yielded up to whoever needs them without any buffering.
1549
# This is the bootstrap phase
1550
if not self._old_root_keys:
1551
# With no old_root_keys we can just shortcut and be ready
1552
# for _flush_new_queue
1553
self._new_queue = list(self._new_root_keys)
1555
old_chks_to_enqueue = self._read_old_roots()
1556
# filter out any root keys that are already known to be uninteresting
1557
new_keys = set(self._new_root_keys).difference(self._all_old_chks)
1558
# These are prefixes that are present in new_keys that we are
1560
new_prefixes = set()
1561
# We are about to yield all of these, so we don't want them getting
1562
# added a second time
1563
processed_new_refs = self._processed_new_refs
1564
processed_new_refs.update(new_keys)
1565
for record, node, prefix_refs, items in \
1566
self._read_nodes_from_store(new_keys):
1567
# At this level, we now know all the uninteresting references
1568
# So we filter and queue up whatever is remaining
1569
prefix_refs = [p_r for p_r in prefix_refs
1570
if p_r[1] not in self._all_old_chks
1571
and p_r[1] not in processed_new_refs]
1572
refs = [p_r[1] for p_r in prefix_refs]
1573
new_prefixes.update([p_r[0] for p_r in prefix_refs])
1574
self._new_queue.extend(refs)
1575
# TODO: We can potentially get multiple items here, however the
1576
# current design allows for this, as callers will do the work
1577
# to make the results unique. We might profile whether we
1578
# gain anything by ensuring unique return values for items
1579
new_items = [item for item in items
1580
if item not in self._all_old_items]
1581
self._new_item_queue.extend(new_items)
1582
new_prefixes.update([self._search_key_func(item[0])
1583
for item in new_items])
1584
processed_new_refs.update(refs)
1586
# For new_prefixes we have the full length prefixes queued up.
1587
# However, we also need possible prefixes. (If we have a known ref to
1588
# 'ab', then we also need to include 'a'.) So expand the
1589
# new_prefixes to include all shorter prefixes
1590
for prefix in list(new_prefixes):
1591
new_prefixes.update([prefix[:i] for i in xrange(1, len(prefix))])
1592
self._enqueue_old(new_prefixes, old_chks_to_enqueue)
1594
def _flush_new_queue(self):
1595
# No need to maintain the heap invariant anymore, just pull things out
1597
refs = set(self._new_queue)
1598
self._new_queue = []
1599
# First pass, flush all interesting items and convert to using direct refs
1600
all_old_chks = self._all_old_chks
1601
processed_new_refs = self._processed_new_refs
1602
all_old_items = self._all_old_items
1603
new_items = [item for item in self._new_item_queue
1604
if item not in all_old_items]
1605
self._new_item_queue = []
1607
yield None, new_items
1608
refs = refs.difference(all_old_chks)
1611
next_refs_update = next_refs.update
1612
# Inlining _read_nodes_from_store improves 'bzr branch bzr.dev'
1613
# from 1m54s to 1m51s. Consider it.
1614
for record, _, p_refs, items in self._read_nodes_from_store(refs):
1615
items = [item for item in items
1616
if item not in all_old_items]
1618
next_refs_update([p_r[1] for p_r in p_refs])
1619
next_refs = next_refs.difference(all_old_chks)
1620
next_refs = next_refs.difference(processed_new_refs)
1621
processed_new_refs.update(next_refs)
1624
def _process_next_old(self):
1625
# Since we don't filter uninteresting any further than during
1626
# _read_all_roots, process the whole queue in a single pass.
1627
refs = self._old_queue
1628
self._old_queue = []
1629
all_old_chks = self._all_old_chks
1630
for record, _, prefix_refs, items in self._read_nodes_from_store(refs):
1631
self._all_old_items.update(items)
1632
refs = [r for _,r in prefix_refs if r not in all_old_chks]
1633
self._old_queue.extend(refs)
1634
all_old_chks.update(refs)
1636
def _process_queues(self):
1637
while self._old_queue:
1638
self._process_next_old()
1639
return self._flush_new_queue()
1642
for record in self._read_all_roots():
1644
for record, items in self._process_queues():
1648
def iter_interesting_nodes(store, interesting_root_keys,
1649
uninteresting_root_keys, pb=None):
1650
"""Given root keys, find interesting nodes.
1652
Evaluate nodes referenced by interesting_root_keys. Ones that are also
1653
referenced from uninteresting_root_keys are not considered interesting.
1655
:param interesting_root_keys: keys which should be part of the
1656
"interesting" nodes (which will be yielded)
1657
:param uninteresting_root_keys: keys which should be filtered out of the
1660
(interesting record, {interesting key:values})
1662
iterator = CHKMapDifference(store, interesting_root_keys,
1663
uninteresting_root_keys,
1664
search_key_func=store._search_key_func,
1666
return iterator.process()
1670
from bzrlib._chk_map_pyx import (
1673
_deserialise_leaf_node,
1674
_deserialise_internal_node,
1676
except ImportError, e:
1677
osutils.failed_to_load_extension(e)
1678
from bzrlib._chk_map_py import (
1681
_deserialise_leaf_node,
1682
_deserialise_internal_node,
1684
search_key_registry.register('hash-16-way', _search_key_16)
1685
search_key_registry.register('hash-255-way', _search_key_255)
1687
def _check_key(key):
1688
if type(key) is not StaticTuple:
1689
raise TypeError('key %r is not StaticTuple but %s' % (key, type(key)))
1691
raise ValueError('key %r should have length 1, not %d' % (key, len(key),))
1692
if type(key[0]) is not str:
1693
raise TypeError('key %r should hold a str, not %r'
1694
% (key, type(key[0])))
1695
if not key[0].startswith('sha1:'):
1696
raise ValueError('key %r should point to a sha1:' % (key,))
added
removed
bzrlib/chk_map.py
