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
if type(root_key) is not StaticTuple:
226
raise AssertionError('we got a %s instead of a StaticTuple'
231
def _create_via_map(klass, store, initial_value, maximum_size=0,
232
key_width=1, search_key_func=None):
233
result = klass(store, None, search_key_func=search_key_func)
234
result._root_node.set_maximum_size(maximum_size)
235
result._root_node._key_width = key_width
237
for key, value in initial_value.items():
238
delta.append((None, key, value))
239
root_key = result.apply_delta(delta)
243
def _create_directly(klass, store, initial_value, maximum_size=0,
244
key_width=1, search_key_func=None):
245
node = LeafNode(search_key_func=search_key_func)
246
node.set_maximum_size(maximum_size)
247
node._key_width = key_width
248
as_st = StaticTuple.from_sequence
249
node._items = dict([(as_st(key), val) for key, val
250
in initial_value.iteritems()])
251
node._raw_size = sum([node._key_value_len(key, value)
252
for key,value in node._items.iteritems()])
253
node._len = len(node._items)
254
node._compute_search_prefix()
255
node._compute_serialised_prefix()
258
and node._current_size() > maximum_size):
259
prefix, node_details = node._split(store)
260
if len(node_details) == 1:
261
raise AssertionError('Failed to split using node._split')
262
node = InternalNode(prefix, search_key_func=search_key_func)
263
node.set_maximum_size(maximum_size)
264
node._key_width = key_width
265
for split, subnode in node_details:
266
node.add_node(split, subnode)
267
keys = list(node.serialise(store))
270
def iter_changes(self, basis):
271
"""Iterate over the changes between basis and self.
273
:return: An iterator of tuples: (key, old_value, new_value). Old_value
274
is None for keys only in self; new_value is None for keys only in
278
# Read both trees in lexographic, highest-first order.
279
# Any identical nodes we skip
280
# Any unique prefixes we output immediately.
281
# values in a leaf node are treated as single-value nodes in the tree
282
# which allows them to be not-special-cased. We know to output them
283
# because their value is a string, not a key(tuple) or node.
285
# corner cases to beware of when considering this function:
286
# *) common references are at different heights.
287
# consider two trees:
288
# {'a': LeafNode={'aaa':'foo', 'aab':'bar'}, 'b': LeafNode={'b'}}
289
# {'a': InternalNode={'aa':LeafNode={'aaa':'foo', 'aab':'bar'},
290
# 'ab':LeafNode={'ab':'bar'}}
291
# 'b': LeafNode={'b'}}
292
# the node with aaa/aab will only be encountered in the second tree
293
# after reading the 'a' subtree, but it is encountered in the first
294
# tree immediately. Variations on this may have read internal nodes
295
# like this. we want to cut the entire pending subtree when we
296
# realise we have a common node. For this we use a list of keys -
297
# the path to a node - and check the entire path is clean as we
299
if self._node_key(self._root_node) == self._node_key(basis._root_node):
303
excluded_keys = set()
304
self_node = self._root_node
305
basis_node = basis._root_node
306
# A heap, each element is prefix, node(tuple/NodeObject/string),
307
# key_path (a list of tuples, tail-sharing down the tree.)
310
def process_node(node, path, a_map, pending):
311
# take a node and expand it
312
node = a_map._get_node(node)
313
if type(node) == LeafNode:
314
path = (node._key, path)
315
for key, value in node._items.items():
316
# For a LeafNode, the key is a serialized_key, rather than
317
# a search_key, but the heap is using search_keys
318
search_key = node._search_key_func(key)
319
heapq.heappush(pending, (search_key, key, value, path))
321
# type(node) == InternalNode
322
path = (node._key, path)
323
for prefix, child in node._items.items():
324
heapq.heappush(pending, (prefix, None, child, path))
325
def process_common_internal_nodes(self_node, basis_node):
326
self_items = set(self_node._items.items())
327
basis_items = set(basis_node._items.items())
328
path = (self_node._key, None)
329
for prefix, child in self_items - basis_items:
330
heapq.heappush(self_pending, (prefix, None, child, path))
331
path = (basis_node._key, None)
332
for prefix, child in basis_items - self_items:
333
heapq.heappush(basis_pending, (prefix, None, child, path))
334
def process_common_leaf_nodes(self_node, basis_node):
335
self_items = set(self_node._items.items())
336
basis_items = set(basis_node._items.items())
337
path = (self_node._key, None)
338
for key, value in self_items - basis_items:
339
prefix = self._search_key_func(key)
340
heapq.heappush(self_pending, (prefix, key, value, path))
341
path = (basis_node._key, None)
342
for key, value in basis_items - self_items:
343
prefix = basis._search_key_func(key)
344
heapq.heappush(basis_pending, (prefix, key, value, path))
345
def process_common_prefix_nodes(self_node, self_path,
346
basis_node, basis_path):
347
# Would it be more efficient if we could request both at the same
349
self_node = self._get_node(self_node)
350
basis_node = basis._get_node(basis_node)
351
if (type(self_node) == InternalNode
352
and type(basis_node) == InternalNode):
353
# Matching internal nodes
354
process_common_internal_nodes(self_node, basis_node)
355
elif (type(self_node) == LeafNode
356
and type(basis_node) == LeafNode):
357
process_common_leaf_nodes(self_node, basis_node)
359
process_node(self_node, self_path, self, self_pending)
360
process_node(basis_node, basis_path, basis, basis_pending)
361
process_common_prefix_nodes(self_node, None, basis_node, None)
364
excluded_keys = set()
365
def check_excluded(key_path):
366
# Note that this is N^2, it depends on us trimming trees
367
# aggressively to not become slow.
368
# A better implementation would probably have a reverse map
369
# back to the children of a node, and jump straight to it when
370
# a common node is detected, the proceed to remove the already
371
# pending children. bzrlib.graph has a searcher module with a
373
while key_path is not None:
374
key, key_path = key_path
375
if key in excluded_keys:
380
while self_pending or basis_pending:
383
# self is exhausted: output remainder of basis
384
for prefix, key, node, path in basis_pending:
385
if check_excluded(path):
387
node = basis._get_node(node)
390
yield (key, node, None)
392
# subtree - fastpath the entire thing.
393
for key, value in node.iteritems(basis._store):
394
yield (key, value, None)
396
elif not basis_pending:
397
# basis is exhausted: output remainder of self.
398
for prefix, key, node, path in self_pending:
399
if check_excluded(path):
401
node = self._get_node(node)
404
yield (key, None, node)
406
# subtree - fastpath the entire thing.
407
for key, value in node.iteritems(self._store):
408
yield (key, None, value)
411
# XXX: future optimisation - yield the smaller items
412
# immediately rather than pushing everything on/off the
413
# heaps. Applies to both internal nodes and leafnodes.
414
if self_pending[0][0] < basis_pending[0][0]:
416
prefix, key, node, path = heapq.heappop(self_pending)
417
if check_excluded(path):
421
yield (key, None, node)
423
process_node(node, path, self, self_pending)
425
elif self_pending[0][0] > basis_pending[0][0]:
427
prefix, key, node, path = heapq.heappop(basis_pending)
428
if check_excluded(path):
432
yield (key, node, None)
434
process_node(node, path, basis, basis_pending)
437
# common prefix: possibly expand both
438
if self_pending[0][1] is None:
443
if basis_pending[0][1] is None:
448
if not read_self and not read_basis:
449
# compare a common value
450
self_details = heapq.heappop(self_pending)
451
basis_details = heapq.heappop(basis_pending)
452
if self_details[2] != basis_details[2]:
453
yield (self_details[1],
454
basis_details[2], self_details[2])
456
# At least one side wasn't a simple value
457
if (self._node_key(self_pending[0][2]) ==
458
self._node_key(basis_pending[0][2])):
459
# Identical pointers, skip (and don't bother adding to
460
# excluded, it won't turn up again.
461
heapq.heappop(self_pending)
462
heapq.heappop(basis_pending)
464
# Now we need to expand this node before we can continue
465
if read_self and read_basis:
466
# Both sides start with the same prefix, so process
468
self_prefix, _, self_node, self_path = heapq.heappop(
470
basis_prefix, _, basis_node, basis_path = heapq.heappop(
472
if self_prefix != basis_prefix:
473
raise AssertionError(
474
'%r != %r' % (self_prefix, basis_prefix))
475
process_common_prefix_nodes(
476
self_node, self_path,
477
basis_node, basis_path)
480
prefix, key, node, path = heapq.heappop(self_pending)
481
if check_excluded(path):
483
process_node(node, path, self, self_pending)
485
prefix, key, node, path = heapq.heappop(basis_pending)
486
if check_excluded(path):
488
process_node(node, path, basis, basis_pending)
491
def iteritems(self, key_filter=None):
492
"""Iterate over the entire CHKMap's contents."""
494
if key_filter is not None:
495
as_st = StaticTuple.from_sequence
496
key_filter = [as_st(key) for key in key_filter]
497
return self._root_node.iteritems(self._store, key_filter=key_filter)
500
"""Return the key for this map."""
501
if type(self._root_node) is StaticTuple:
502
return self._root_node
504
return self._root_node._key
508
return len(self._root_node)
510
def map(self, key, value):
511
"""Map a key tuple to value.
513
:param key: A key to map.
514
:param value: The value to assign to key.
516
key = StaticTuple.from_sequence(key)
517
# Need a root object.
519
prefix, node_details = self._root_node.map(self._store, key, value)
520
if len(node_details) == 1:
521
self._root_node = node_details[0][1]
523
self._root_node = InternalNode(prefix,
524
search_key_func=self._search_key_func)
525
self._root_node.set_maximum_size(node_details[0][1].maximum_size)
526
self._root_node._key_width = node_details[0][1]._key_width
527
for split, node in node_details:
528
self._root_node.add_node(split, node)
530
def _node_key(self, node):
531
"""Get the key for a node whether it's a tuple or node."""
532
if type(node) is tuple:
533
node = StaticTuple.from_sequence(node)
534
if type(node) is StaticTuple:
539
def unmap(self, key, check_remap=True):
540
"""remove key from the map."""
541
key = StaticTuple.from_sequence(key)
543
if type(self._root_node) is InternalNode:
544
unmapped = self._root_node.unmap(self._store, key,
545
check_remap=check_remap)
547
unmapped = self._root_node.unmap(self._store, key)
548
self._root_node = unmapped
550
def _check_remap(self):
551
"""Check if nodes can be collapsed."""
553
if type(self._root_node) is InternalNode:
554
self._root_node._check_remap(self._store)
557
"""Save the map completely.
559
:return: The key of the root node.
561
if type(self._root_node) is StaticTuple:
563
return self._root_node
564
keys = list(self._root_node.serialise(self._store))
569
"""Base class defining the protocol for CHK Map nodes.
571
:ivar _raw_size: The total size of the serialized key:value data, before
572
adding the header bytes, and without prefix compression.
575
__slots__ = ('_key', '_len', '_maximum_size', '_key_width',
576
'_raw_size', '_items', '_search_prefix', '_search_key_func'
579
def __init__(self, key_width=1):
582
:param key_width: The width of keys for this node.
585
# Current number of elements
587
self._maximum_size = 0
588
self._key_width = key_width
589
# current size in bytes
591
# The pointers/values this node has - meaning defined by child classes.
593
# The common search prefix
594
self._search_prefix = None
597
items_str = str(sorted(self._items))
598
if len(items_str) > 20:
599
items_str = items_str[:16] + '...]'
600
return '%s(key:%s len:%s size:%s max:%s prefix:%s items:%s)' % (
601
self.__class__.__name__, self._key, self._len, self._raw_size,
602
self._maximum_size, self._search_prefix, items_str)
611
def maximum_size(self):
612
"""What is the upper limit for adding references to a node."""
613
return self._maximum_size
615
def set_maximum_size(self, new_size):
616
"""Set the size threshold for nodes.
618
:param new_size: The size at which no data is added to a node. 0 for
621
self._maximum_size = new_size
624
def common_prefix(cls, prefix, key):
625
"""Given 2 strings, return the longest prefix common to both.
627
:param prefix: This has been the common prefix for other keys, so it is
628
more likely to be the common prefix in this case as well.
629
:param key: Another string to compare to
631
if key.startswith(prefix):
634
# Is there a better way to do this?
635
for pos, (left, right) in enumerate(zip(prefix, key)):
639
common = prefix[:pos+1]
643
def common_prefix_for_keys(cls, keys):
644
"""Given a list of keys, find their common prefix.
646
:param keys: An iterable of strings.
647
:return: The longest common prefix of all keys.
651
if common_prefix is None:
654
common_prefix = cls.common_prefix(common_prefix, key)
655
if not common_prefix:
656
# if common_prefix is the empty string, then we know it won't
662
# Singleton indicating we have not computed _search_prefix yet
665
class LeafNode(Node):
666
"""A node containing actual key:value pairs.
668
:ivar _items: A dict of key->value items. The key is in tuple form.
669
:ivar _size: The number of bytes that would be used by serializing all of
673
__slots__ = ('_common_serialised_prefix', '_serialise_key')
675
def __init__(self, search_key_func=None):
677
# All of the keys in this leaf node share this common prefix
678
self._common_serialised_prefix = None
679
self._serialise_key = '\x00'.join
680
if search_key_func is None:
681
self._search_key_func = _search_key_plain
683
self._search_key_func = search_key_func
686
items_str = str(sorted(self._items))
687
if len(items_str) > 20:
688
items_str = items_str[:16] + '...]'
690
'%s(key:%s len:%s size:%s max:%s prefix:%s keywidth:%s items:%s)' \
691
% (self.__class__.__name__, self._key, self._len, self._raw_size,
692
self._maximum_size, self._search_prefix, self._key_width, items_str)
694
def _current_size(self):
695
"""Answer the current serialised size of this node.
697
This differs from self._raw_size in that it includes the bytes used for
700
if self._common_serialised_prefix is None:
704
# We will store a single string with the common prefix
705
# And then that common prefix will not be stored in any of the
707
prefix_len = len(self._common_serialised_prefix)
708
bytes_for_items = (self._raw_size - (prefix_len * self._len))
709
return (9 # 'chkleaf:\n'
710
+ len(str(self._maximum_size)) + 1
711
+ len(str(self._key_width)) + 1
712
+ len(str(self._len)) + 1
717
def deserialise(klass, bytes, key, search_key_func=None):
718
"""Deserialise bytes, with key key, into a LeafNode.
720
:param bytes: The bytes of the node.
721
:param key: The key that the serialised node has.
723
key = static_tuple.expect_static_tuple(key)
724
return _deserialise_leaf_node(bytes, key,
725
search_key_func=search_key_func)
727
def iteritems(self, store, key_filter=None):
728
"""Iterate over items in the node.
730
:param key_filter: A filter to apply to the node. It should be a
731
list/set/dict or similar repeatedly iterable container.
733
if key_filter is not None:
734
# Adjust the filter - short elements go to a prefix filter. All
735
# other items are looked up directly.
736
# XXX: perhaps defaultdict? Profiling<rinse and repeat>
738
for key in key_filter:
739
if len(key) == self._key_width:
740
# This filter is meant to match exactly one key, yield it
743
yield key, self._items[key]
745
# This key is not present in this map, continue
748
# Short items, we need to match based on a prefix
749
length_filter = filters.setdefault(len(key), set())
750
length_filter.add(key)
752
filters = filters.items()
753
for item in self._items.iteritems():
754
for length, length_filter in filters:
755
if item[0][:length] in length_filter:
759
for item in self._items.iteritems():
762
def _key_value_len(self, key, value):
763
# TODO: Should probably be done without actually joining the key, but
764
# then that can be done via the C extension
765
return (len(self._serialise_key(key)) + 1
766
+ len(str(value.count('\n'))) + 1
769
def _search_key(self, key):
770
return self._search_key_func(key)
772
def _map_no_split(self, key, value):
773
"""Map a key to a value.
775
This assumes either the key does not already exist, or you have already
776
removed its size and length from self.
778
:return: True if adding this node should cause us to split.
780
self._items[key] = value
781
self._raw_size += self._key_value_len(key, value)
783
serialised_key = self._serialise_key(key)
784
if self._common_serialised_prefix is None:
785
self._common_serialised_prefix = serialised_key
787
self._common_serialised_prefix = self.common_prefix(
788
self._common_serialised_prefix, serialised_key)
789
search_key = self._search_key(key)
790
if self._search_prefix is _unknown:
791
self._compute_search_prefix()
792
if self._search_prefix is None:
793
self._search_prefix = search_key
795
self._search_prefix = self.common_prefix(
796
self._search_prefix, search_key)
798
and self._maximum_size
799
and self._current_size() > self._maximum_size):
800
# Check to see if all of the search_keys for this node are
801
# identical. We allow the node to grow under that circumstance
802
# (we could track this as common state, but it is infrequent)
803
if (search_key != self._search_prefix
804
or not self._are_search_keys_identical()):
808
def _split(self, store):
809
"""We have overflowed.
811
Split this node into multiple LeafNodes, return it up the stack so that
812
the next layer creates a new InternalNode and references the new nodes.
814
:return: (common_serialised_prefix, [(node_serialised_prefix, node)])
816
if self._search_prefix is _unknown:
817
raise AssertionError('Search prefix must be known')
818
common_prefix = self._search_prefix
819
split_at = len(common_prefix) + 1
821
for key, value in self._items.iteritems():
822
search_key = self._search_key(key)
823
prefix = search_key[:split_at]
824
# TODO: Generally only 1 key can be exactly the right length,
825
# which means we can only have 1 key in the node pointed
826
# at by the 'prefix\0' key. We might want to consider
827
# folding it into the containing InternalNode rather than
828
# having a fixed length-1 node.
829
# Note this is probably not true for hash keys, as they
830
# may get a '\00' node anywhere, but won't have keys of
832
if len(prefix) < split_at:
833
prefix += '\x00'*(split_at - len(prefix))
834
if prefix not in result:
835
node = LeafNode(search_key_func=self._search_key_func)
836
node.set_maximum_size(self._maximum_size)
837
node._key_width = self._key_width
838
result[prefix] = node
840
node = result[prefix]
841
sub_prefix, node_details = node.map(store, key, value)
842
if len(node_details) > 1:
843
if prefix != sub_prefix:
844
# This node has been split and is now found via a different
847
new_node = InternalNode(sub_prefix,
848
search_key_func=self._search_key_func)
849
new_node.set_maximum_size(self._maximum_size)
850
new_node._key_width = self._key_width
851
for split, node in node_details:
852
new_node.add_node(split, node)
853
result[prefix] = new_node
854
return common_prefix, result.items()
856
def map(self, store, key, value):
857
"""Map key to value."""
858
if key in self._items:
859
self._raw_size -= self._key_value_len(key, self._items[key])
862
if self._map_no_split(key, value):
863
return self._split(store)
865
if self._search_prefix is _unknown:
866
raise AssertionError('%r must be known' % self._search_prefix)
867
return self._search_prefix, [("", self)]
869
def serialise(self, store):
870
"""Serialise the LeafNode to store.
872
:param store: A VersionedFiles honouring the CHK extensions.
873
:return: An iterable of the keys inserted by this operation.
875
lines = ["chkleaf:\n"]
876
lines.append("%d\n" % self._maximum_size)
877
lines.append("%d\n" % self._key_width)
878
lines.append("%d\n" % self._len)
879
if self._common_serialised_prefix is None:
881
if len(self._items) != 0:
882
raise AssertionError('If _common_serialised_prefix is None'
883
' we should have no items')
885
lines.append('%s\n' % (self._common_serialised_prefix,))
886
prefix_len = len(self._common_serialised_prefix)
887
for key, value in sorted(self._items.items()):
888
# Always add a final newline
889
value_lines = osutils.chunks_to_lines([value + '\n'])
890
serialized = "%s\x00%s\n" % (self._serialise_key(key),
892
if not serialized.startswith(self._common_serialised_prefix):
893
raise AssertionError('We thought the common prefix was %r'
894
' but entry %r does not have it in common'
895
% (self._common_serialised_prefix, serialized))
896
lines.append(serialized[prefix_len:])
897
lines.extend(value_lines)
898
sha1, _, _ = store.add_lines((None,), (), lines)
899
self._key = StaticTuple("sha1:" + sha1,).intern()
900
bytes = ''.join(lines)
901
if len(bytes) != self._current_size():
902
raise AssertionError('Invalid _current_size')
903
_page_cache.add(self._key, bytes)
907
"""Return the references to other CHK's held by this node."""
910
def _compute_search_prefix(self):
911
"""Determine the common search prefix for all keys in this node.
913
:return: A bytestring of the longest search key prefix that is
914
unique within this node.
916
search_keys = [self._search_key_func(key) for key in self._items]
917
self._search_prefix = self.common_prefix_for_keys(search_keys)
918
return self._search_prefix
920
def _are_search_keys_identical(self):
921
"""Check to see if the search keys for all entries are the same.
923
When using a hash as the search_key it is possible for non-identical
924
keys to collide. If that happens enough, we may try overflow a
925
LeafNode, but as all are collisions, we must not split.
927
common_search_key = None
928
for key in self._items:
929
search_key = self._search_key(key)
930
if common_search_key is None:
931
common_search_key = search_key
932
elif search_key != common_search_key:
936
def _compute_serialised_prefix(self):
937
"""Determine the common prefix for serialised keys in this node.
939
:return: A bytestring of the longest serialised key prefix that is
940
unique within this node.
942
serialised_keys = [self._serialise_key(key) for key in self._items]
943
self._common_serialised_prefix = self.common_prefix_for_keys(
945
return self._common_serialised_prefix
947
def unmap(self, store, key):
948
"""Unmap key from the node."""
950
self._raw_size -= self._key_value_len(key, self._items[key])
952
trace.mutter("key %s not found in %r", key, self._items)
957
# Recompute from scratch
958
self._compute_search_prefix()
959
self._compute_serialised_prefix()
963
class InternalNode(Node):
964
"""A node that contains references to other nodes.
966
An InternalNode is responsible for mapping search key prefixes to child
969
:ivar _items: serialised_key => node dictionary. node may be a tuple,
970
LeafNode or InternalNode.
973
__slots__ = ('_node_width',)
975
def __init__(self, prefix='', search_key_func=None):
977
# The size of an internalnode with default values and no children.
978
# How many octets key prefixes within this node are.
980
self._search_prefix = prefix
981
if search_key_func is None:
982
self._search_key_func = _search_key_plain
984
self._search_key_func = search_key_func
986
def add_node(self, prefix, node):
987
"""Add a child node with prefix prefix, and node node.
989
:param prefix: The search key prefix for node.
990
:param node: The node being added.
992
if self._search_prefix is None:
993
raise AssertionError("_search_prefix should not be None")
994
if not prefix.startswith(self._search_prefix):
995
raise AssertionError("prefixes mismatch: %s must start with %s"
996
% (prefix,self._search_prefix))
997
if len(prefix) != len(self._search_prefix) + 1:
998
raise AssertionError("prefix wrong length: len(%s) is not %d" %
999
(prefix, len(self._search_prefix) + 1))
1000
self._len += len(node)
1001
if not len(self._items):
1002
self._node_width = len(prefix)
1003
if self._node_width != len(self._search_prefix) + 1:
1004
raise AssertionError("node width mismatch: %d is not %d" %
1005
(self._node_width, len(self._search_prefix) + 1))
1006
self._items[prefix] = node
1009
def _current_size(self):
1010
"""Answer the current serialised size of this node."""
1011
return (self._raw_size + len(str(self._len)) + len(str(self._key_width)) +
1012
len(str(self._maximum_size)))
1015
def deserialise(klass, bytes, key, search_key_func=None):
1016
"""Deserialise bytes to an InternalNode, with key key.
1018
:param bytes: The bytes of the node.
1019
:param key: The key that the serialised node has.
1020
:return: An InternalNode instance.
1022
key = static_tuple.expect_static_tuple(key)
1023
return _deserialise_internal_node(bytes, key,
1024
search_key_func=search_key_func)
1026
def iteritems(self, store, key_filter=None):
1027
for node, node_filter in self._iter_nodes(store, key_filter=key_filter):
1028
for item in node.iteritems(store, key_filter=node_filter):
1031
def _iter_nodes(self, store, key_filter=None, batch_size=None):
1032
"""Iterate over node objects which match key_filter.
1034
:param store: A store to use for accessing content.
1035
:param key_filter: A key filter to filter nodes. Only nodes that might
1036
contain a key in key_filter will be returned.
1037
:param batch_size: If not None, then we will return the nodes that had
1038
to be read using get_record_stream in batches, rather than reading
1040
:return: An iterable of nodes. This function does not have to be fully
1041
consumed. (There will be no pending I/O when items are being returned.)
1043
# Map from chk key ('sha1:...',) to (prefix, key_filter)
1044
# prefix is the key in self._items to use, key_filter is the key_filter
1045
# entries that would match this node
1048
if key_filter is None:
1049
# yielding all nodes, yield whatever we have, and queue up a read
1050
# for whatever we are missing
1052
for prefix, node in self._items.iteritems():
1053
if node.__class__ is StaticTuple:
1054
keys[node] = (prefix, None)
1057
elif len(key_filter) == 1:
1058
# Technically, this path could also be handled by the first check
1059
# in 'self._node_width' in length_filters. However, we can handle
1060
# this case without spending any time building up the
1061
# prefix_to_keys, etc state.
1063
# This is a bit ugly, but TIMEIT showed it to be by far the fastest
1064
# 0.626us list(key_filter)[0]
1065
# is a func() for list(), 2 mallocs, and a getitem
1066
# 0.489us [k for k in key_filter][0]
1067
# still has the mallocs, avoids the func() call
1068
# 0.350us iter(key_filter).next()
1069
# has a func() call, and mallocs an iterator
1070
# 0.125us for key in key_filter: pass
1071
# no func() overhead, might malloc an iterator
1072
# 0.105us for key in key_filter: break
1073
# no func() overhead, might malloc an iterator, probably
1074
# avoids checking an 'else' clause as part of the for
1075
for key in key_filter:
1077
search_prefix = self._search_prefix_filter(key)
1078
if len(search_prefix) == self._node_width:
1079
# This item will match exactly, so just do a dict lookup, and
1080
# see what we can return
1083
node = self._items[search_prefix]
1085
# A given key can only match 1 child node, if it isn't
1086
# there, then we can just return nothing
1088
if node.__class__ is StaticTuple:
1089
keys[node] = (search_prefix, [key])
1091
# This is loaded, and the only thing that can match,
1096
# First, convert all keys into a list of search prefixes
1097
# Aggregate common prefixes, and track the keys they come from
1100
for key in key_filter:
1101
search_prefix = self._search_prefix_filter(key)
1102
length_filter = length_filters.setdefault(
1103
len(search_prefix), set())
1104
length_filter.add(search_prefix)
1105
prefix_to_keys.setdefault(search_prefix, []).append(key)
1107
if (self._node_width in length_filters
1108
and len(length_filters) == 1):
1109
# all of the search prefixes match exactly _node_width. This
1110
# means that everything is an exact match, and we can do a
1111
# lookup into self._items, rather than iterating over the items
1113
search_prefixes = length_filters[self._node_width]
1114
for search_prefix in search_prefixes:
1116
node = self._items[search_prefix]
1118
# We can ignore this one
1120
node_key_filter = prefix_to_keys[search_prefix]
1121
if node.__class__ is StaticTuple:
1122
keys[node] = (search_prefix, node_key_filter)
1124
yield node, node_key_filter
1126
# The slow way. We walk every item in self._items, and check to
1127
# see if there are any matches
1128
length_filters = length_filters.items()
1129
for prefix, node in self._items.iteritems():
1130
node_key_filter = []
1131
for length, length_filter in length_filters:
1132
sub_prefix = prefix[:length]
1133
if sub_prefix in length_filter:
1134
node_key_filter.extend(prefix_to_keys[sub_prefix])
1135
if node_key_filter: # this key matched something, yield it
1136
if node.__class__ is StaticTuple:
1137
keys[node] = (prefix, node_key_filter)
1139
yield node, node_key_filter
1141
# Look in the page cache for some more bytes
1145
bytes = _page_cache[key]
1149
node = _deserialise(bytes, key,
1150
search_key_func=self._search_key_func)
1151
prefix, node_key_filter = keys[key]
1152
self._items[prefix] = node
1154
yield node, node_key_filter
1155
for key in found_keys:
1158
# demand load some pages.
1159
if batch_size is None:
1160
# Read all the keys in
1161
batch_size = len(keys)
1162
key_order = list(keys)
1163
for batch_start in range(0, len(key_order), batch_size):
1164
batch = key_order[batch_start:batch_start + batch_size]
1165
# We have to fully consume the stream so there is no pending
1166
# I/O, so we buffer the nodes for now.
1167
stream = store.get_record_stream(batch, 'unordered', True)
1168
node_and_filters = []
1169
for record in stream:
1170
bytes = record.get_bytes_as('fulltext')
1171
node = _deserialise(bytes, record.key,
1172
search_key_func=self._search_key_func)
1173
prefix, node_key_filter = keys[record.key]
1174
node_and_filters.append((node, node_key_filter))
1175
self._items[prefix] = node
1176
_page_cache.add(record.key, bytes)
1177
for info in node_and_filters:
1180
def map(self, store, key, value):
1181
"""Map key to value."""
1182
if not len(self._items):
1183
raise AssertionError("can't map in an empty InternalNode.")
1184
search_key = self._search_key(key)
1185
if self._node_width != len(self._search_prefix) + 1:
1186
raise AssertionError("node width mismatch: %d is not %d" %
1187
(self._node_width, len(self._search_prefix) + 1))
1188
if not search_key.startswith(self._search_prefix):
1189
# This key doesn't fit in this index, so we need to split at the
1190
# point where it would fit, insert self into that internal node,
1191
# and then map this key into that node.
1192
new_prefix = self.common_prefix(self._search_prefix,
1194
new_parent = InternalNode(new_prefix,
1195
search_key_func=self._search_key_func)
1196
new_parent.set_maximum_size(self._maximum_size)
1197
new_parent._key_width = self._key_width
1198
new_parent.add_node(self._search_prefix[:len(new_prefix)+1],
1200
return new_parent.map(store, key, value)
1201
children = [node for node, _
1202
in self._iter_nodes(store, key_filter=[key])]
1207
child = self._new_child(search_key, LeafNode)
1208
old_len = len(child)
1209
if type(child) is LeafNode:
1210
old_size = child._current_size()
1213
prefix, node_details = child.map(store, key, value)
1214
if len(node_details) == 1:
1215
# child may have shrunk, or might be a new node
1216
child = node_details[0][1]
1217
self._len = self._len - old_len + len(child)
1218
self._items[search_key] = child
1221
if type(child) is LeafNode:
1222
if old_size is None:
1223
# The old node was an InternalNode which means it has now
1224
# collapsed, so we need to check if it will chain to a
1225
# collapse at this level.
1226
trace.mutter("checking remap as InternalNode -> LeafNode")
1227
new_node = self._check_remap(store)
1229
# If the LeafNode has shrunk in size, we may want to run
1230
# a remap check. Checking for a remap is expensive though
1231
# and the frequency of a successful remap is very low.
1232
# Shrinkage by small amounts is common, so we only do the
1233
# remap check if the new_size is low or the shrinkage
1234
# amount is over a configurable limit.
1235
new_size = child._current_size()
1236
shrinkage = old_size - new_size
1237
if (shrinkage > 0 and new_size < _INTERESTING_NEW_SIZE
1238
or shrinkage > _INTERESTING_SHRINKAGE_LIMIT):
1240
"checking remap as size shrunk by %d to be %d",
1241
shrinkage, new_size)
1242
new_node = self._check_remap(store)
1243
if new_node._search_prefix is None:
1244
raise AssertionError("_search_prefix should not be None")
1245
return new_node._search_prefix, [('', new_node)]
1246
# child has overflown - create a new intermediate node.
1247
# XXX: This is where we might want to try and expand our depth
1248
# to refer to more bytes of every child (which would give us
1249
# multiple pointers to child nodes, but less intermediate nodes)
1250
child = self._new_child(search_key, InternalNode)
1251
child._search_prefix = prefix
1252
for split, node in node_details:
1253
child.add_node(split, node)
1254
self._len = self._len - old_len + len(child)
1256
return self._search_prefix, [("", self)]
1258
def _new_child(self, search_key, klass):
1259
"""Create a new child node of type klass."""
1261
child.set_maximum_size(self._maximum_size)
1262
child._key_width = self._key_width
1263
child._search_key_func = self._search_key_func
1264
self._items[search_key] = child
1267
def serialise(self, store):
1268
"""Serialise the node to store.
1270
:param store: A VersionedFiles honouring the CHK extensions.
1271
:return: An iterable of the keys inserted by this operation.
1273
for node in self._items.itervalues():
1274
if type(node) is StaticTuple:
1275
# Never deserialised.
1277
if node._key is not None:
1280
for key in node.serialise(store):
1282
lines = ["chknode:\n"]
1283
lines.append("%d\n" % self._maximum_size)
1284
lines.append("%d\n" % self._key_width)
1285
lines.append("%d\n" % self._len)
1286
if self._search_prefix is None:
1287
raise AssertionError("_search_prefix should not be None")
1288
lines.append('%s\n' % (self._search_prefix,))
1289
prefix_len = len(self._search_prefix)
1290
for prefix, node in sorted(self._items.items()):
1291
if type(node) is StaticTuple:
1295
serialised = "%s\x00%s\n" % (prefix, key)
1296
if not serialised.startswith(self._search_prefix):
1297
raise AssertionError("prefixes mismatch: %s must start with %s"
1298
% (serialised, self._search_prefix))
1299
lines.append(serialised[prefix_len:])
1300
sha1, _, _ = store.add_lines((None,), (), lines)
1301
self._key = StaticTuple("sha1:" + sha1,).intern()
1302
_page_cache.add(self._key, ''.join(lines))
1305
def _search_key(self, key):
1306
"""Return the serialised key for key in this node."""
1307
# search keys are fixed width. All will be self._node_width wide, so we
1309
return (self._search_key_func(key) + '\x00'*self._node_width)[:self._node_width]
1311
def _search_prefix_filter(self, key):
1312
"""Serialise key for use as a prefix filter in iteritems."""
1313
return self._search_key_func(key)[:self._node_width]
1315
def _split(self, offset):
1316
"""Split this node into smaller nodes starting at offset.
1318
:param offset: The offset to start the new child nodes at.
1319
:return: An iterable of (prefix, node) tuples. prefix is a byte
1320
prefix for reaching node.
1322
if offset >= self._node_width:
1323
for node in self._items.values():
1324
for result in node._split(offset):
1327
for key, node in self._items.items():
1331
"""Return the references to other CHK's held by this node."""
1332
if self._key is None:
1333
raise AssertionError("unserialised nodes have no refs.")
1335
for value in self._items.itervalues():
1336
if type(value) is StaticTuple:
1339
refs.append(value.key())
1342
def _compute_search_prefix(self, extra_key=None):
1343
"""Return the unique key prefix for this node.
1345
:return: A bytestring of the longest search key prefix that is
1346
unique within this node.
1348
self._search_prefix = self.common_prefix_for_keys(self._items)
1349
return self._search_prefix
1351
def unmap(self, store, key, check_remap=True):
1352
"""Remove key from this node and it's children."""
1353
if not len(self._items):
1354
raise AssertionError("can't unmap in an empty InternalNode.")
1355
children = [node for node, _
1356
in self._iter_nodes(store, key_filter=[key])]
1362
unmapped = child.unmap(store, key)
1364
search_key = self._search_key(key)
1365
if len(unmapped) == 0:
1366
# All child nodes are gone, remove the child:
1367
del self._items[search_key]
1370
# Stash the returned node
1371
self._items[search_key] = unmapped
1372
if len(self._items) == 1:
1373
# this node is no longer needed:
1374
return self._items.values()[0]
1375
if type(unmapped) is InternalNode:
1378
return self._check_remap(store)
1382
def _check_remap(self, store):
1383
"""Check if all keys contained by children fit in a single LeafNode.
1385
:param store: A store to use for reading more nodes
1386
:return: Either self, or a new LeafNode which should replace self.
1388
# Logic for how we determine when we need to rebuild
1389
# 1) Implicitly unmap() is removing a key which means that the child
1390
# nodes are going to be shrinking by some extent.
1391
# 2) If all children are LeafNodes, it is possible that they could be
1392
# combined into a single LeafNode, which can then completely replace
1393
# this internal node with a single LeafNode
1394
# 3) If *one* child is an InternalNode, we assume it has already done
1395
# all the work to determine that its children cannot collapse, and
1396
# we can then assume that those nodes *plus* the current nodes don't
1397
# have a chance of collapsing either.
1398
# So a very cheap check is to just say if 'unmapped' is an
1399
# InternalNode, we don't have to check further.
1401
# TODO: Another alternative is to check the total size of all known
1402
# LeafNodes. If there is some formula we can use to determine the
1403
# final size without actually having to read in any more
1404
# children, it would be nice to have. However, we have to be
1405
# careful with stuff like nodes that pull out the common prefix
1406
# of each key, as adding a new key can change the common prefix
1407
# and cause size changes greater than the length of one key.
1408
# So for now, we just add everything to a new Leaf until it
1409
# splits, as we know that will give the right answer
1410
new_leaf = LeafNode(search_key_func=self._search_key_func)
1411
new_leaf.set_maximum_size(self._maximum_size)
1412
new_leaf._key_width = self._key_width
1413
# A batch_size of 16 was chosen because:
1414
# a) In testing, a 4k page held 14 times. So if we have more than 16
1415
# leaf nodes we are unlikely to hold them in a single new leaf
1416
# node. This still allows for 1 round trip
1417
# b) With 16-way fan out, we can still do a single round trip
1418
# c) With 255-way fan out, we don't want to read all 255 and destroy
1419
# the page cache, just to determine that we really don't need it.
1420
for node, _ in self._iter_nodes(store, batch_size=16):
1421
if type(node) is InternalNode:
1422
# Without looking at any leaf nodes, we are sure
1424
for key, value in node._items.iteritems():
1425
if new_leaf._map_no_split(key, value):
1427
trace.mutter("remap generated a new LeafNode")
1431
def _deserialise(bytes, key, search_key_func):
1432
"""Helper for repositorydetails - convert bytes to a node."""
1433
if bytes.startswith("chkleaf:\n"):
1434
node = LeafNode.deserialise(bytes, key, search_key_func=search_key_func)
1435
elif bytes.startswith("chknode:\n"):
1436
node = InternalNode.deserialise(bytes, key,
1437
search_key_func=search_key_func)
1439
raise AssertionError("Unknown node type.")
1443
class CHKMapDifference(object):
1444
"""Iterate the stored pages and key,value pairs for (new - old).
1446
This class provides a generator over the stored CHK pages and the
1447
(key, value) pairs that are in any of the new maps and not in any of the
1450
Note that it may yield chk pages that are common (especially root nodes),
1451
but it won't yield (key,value) pairs that are common.
1454
def __init__(self, store, new_root_keys, old_root_keys,
1455
search_key_func, pb=None):
1456
# TODO: Should we add a StaticTuple barrier here? It would be nice to
1457
# force callers to use StaticTuple, because there will often be
1458
# lots of keys passed in here. And even if we cast it locally,
1459
# that just meanst that we will have *both* a StaticTuple and a
1460
# tuple() in memory, referring to the same object. (so a net
1461
# increase in memory, not a decrease.)
1463
self._new_root_keys = new_root_keys
1464
self._old_root_keys = old_root_keys
1466
# All uninteresting chks that we have seen. By the time they are added
1467
# here, they should be either fully ignored, or queued up for
1469
# TODO: This might grow to a large size if there are lots of merge
1470
# parents, etc. However, it probably doesn't scale to O(history)
1471
# like _processed_new_refs does.
1472
self._all_old_chks = set(self._old_root_keys)
1473
# All items that we have seen from the old_root_keys
1474
self._all_old_items = set()
1475
# These are interesting items which were either read, or already in the
1476
# interesting queue (so we don't need to walk them again)
1477
# TODO: processed_new_refs becomes O(all_chks), consider switching to
1479
self._processed_new_refs = set()
1480
self._search_key_func = search_key_func
1482
# The uninteresting and interesting nodes to be searched
1483
self._old_queue = []
1484
self._new_queue = []
1485
# Holds the (key, value) items found when processing the root nodes,
1486
# waiting for the uninteresting nodes to be walked
1487
self._new_item_queue = []
1490
def _read_nodes_from_store(self, keys):
1491
# We chose not to use _page_cache, because we think in terms of records
1492
# to be yielded. Also, we expect to touch each page only 1 time during
1493
# this code. (We may want to evaluate saving the raw bytes into the
1494
# page cache, which would allow a working tree update after the fetch
1495
# to not have to read the bytes again.)
1496
as_st = StaticTuple.from_sequence
1497
stream = self._store.get_record_stream(keys, 'unordered', True)
1498
for record in stream:
1499
if self._pb is not None:
1501
if record.storage_kind == 'absent':
1502
raise errors.NoSuchRevision(self._store, record.key)
1503
bytes = record.get_bytes_as('fulltext')
1504
node = _deserialise(bytes, record.key,
1505
search_key_func=self._search_key_func)
1506
if type(node) is InternalNode:
1507
# Note we don't have to do node.refs() because we know that
1508
# there are no children that have been pushed into this node
1509
# Note: Using as_st() here seemed to save 1.2MB, which would
1510
# indicate that we keep 100k prefix_refs around while
1511
# processing. They *should* be shorter lived than that...
1512
# It does cost us ~10s of processing time
1513
#prefix_refs = [as_st(item) for item in node._items.iteritems()]
1514
prefix_refs = node._items.items()
1518
# Note: We don't use a StaticTuple here. Profiling showed a
1519
# minor memory improvement (0.8MB out of 335MB peak 0.2%)
1520
# But a significant slowdown (15s / 145s, or 10%)
1521
items = node._items.items()
1522
yield record, node, prefix_refs, items
1524
def _read_old_roots(self):
1525
old_chks_to_enqueue = []
1526
all_old_chks = self._all_old_chks
1527
for record, node, prefix_refs, items in \
1528
self._read_nodes_from_store(self._old_root_keys):
1529
# Uninteresting node
1530
prefix_refs = [p_r for p_r in prefix_refs
1531
if p_r[1] not in all_old_chks]
1532
new_refs = [p_r[1] for p_r in prefix_refs]
1533
all_old_chks.update(new_refs)
1534
# TODO: This might be a good time to turn items into StaticTuple
1535
# instances and possibly intern them. However, this does not
1536
# impact 'initial branch' performance, so I'm not worrying
1538
self._all_old_items.update(items)
1539
# Queue up the uninteresting references
1540
# Don't actually put them in the 'to-read' queue until we have
1541
# finished checking the interesting references
1542
old_chks_to_enqueue.extend(prefix_refs)
1543
return old_chks_to_enqueue
1545
def _enqueue_old(self, new_prefixes, old_chks_to_enqueue):
1546
# At this point, we have read all the uninteresting and interesting
1547
# items, so we can queue up the uninteresting stuff, knowing that we've
1548
# handled the interesting ones
1549
for prefix, ref in old_chks_to_enqueue:
1550
not_interesting = True
1551
for i in xrange(len(prefix), 0, -1):
1552
if prefix[:i] in new_prefixes:
1553
not_interesting = False
1556
# This prefix is not part of the remaining 'interesting set'
1558
self._old_queue.append(ref)
1560
def _read_all_roots(self):
1561
"""Read the root pages.
1563
This is structured as a generator, so that the root records can be
1564
yielded up to whoever needs them without any buffering.
1566
# This is the bootstrap phase
1567
if not self._old_root_keys:
1568
# With no old_root_keys we can just shortcut and be ready
1569
# for _flush_new_queue
1570
self._new_queue = list(self._new_root_keys)
1572
old_chks_to_enqueue = self._read_old_roots()
1573
# filter out any root keys that are already known to be uninteresting
1574
new_keys = set(self._new_root_keys).difference(self._all_old_chks)
1575
# These are prefixes that are present in new_keys that we are
1577
new_prefixes = set()
1578
# We are about to yield all of these, so we don't want them getting
1579
# added a second time
1580
processed_new_refs = self._processed_new_refs
1581
processed_new_refs.update(new_keys)
1582
for record, node, prefix_refs, items in \
1583
self._read_nodes_from_store(new_keys):
1584
# At this level, we now know all the uninteresting references
1585
# So we filter and queue up whatever is remaining
1586
prefix_refs = [p_r for p_r in prefix_refs
1587
if p_r[1] not in self._all_old_chks
1588
and p_r[1] not in processed_new_refs]
1589
refs = [p_r[1] for p_r in prefix_refs]
1590
new_prefixes.update([p_r[0] for p_r in prefix_refs])
1591
self._new_queue.extend(refs)
1592
# TODO: We can potentially get multiple items here, however the
1593
# current design allows for this, as callers will do the work
1594
# to make the results unique. We might profile whether we
1595
# gain anything by ensuring unique return values for items
1596
# TODO: This might be a good time to cast to StaticTuple, as
1597
# self._new_item_queue will hold the contents of multiple
1598
# records for an extended lifetime
1599
new_items = [item for item in items
1600
if item not in self._all_old_items]
1601
self._new_item_queue.extend(new_items)
1602
new_prefixes.update([self._search_key_func(item[0])
1603
for item in new_items])
1604
processed_new_refs.update(refs)
1606
# For new_prefixes we have the full length prefixes queued up.
1607
# However, we also need possible prefixes. (If we have a known ref to
1608
# 'ab', then we also need to include 'a'.) So expand the
1609
# new_prefixes to include all shorter prefixes
1610
for prefix in list(new_prefixes):
1611
new_prefixes.update([prefix[:i] for i in xrange(1, len(prefix))])
1612
self._enqueue_old(new_prefixes, old_chks_to_enqueue)
1614
def _flush_new_queue(self):
1615
# No need to maintain the heap invariant anymore, just pull things out
1617
refs = set(self._new_queue)
1618
self._new_queue = []
1619
# First pass, flush all interesting items and convert to using direct refs
1620
all_old_chks = self._all_old_chks
1621
processed_new_refs = self._processed_new_refs
1622
all_old_items = self._all_old_items
1623
new_items = [item for item in self._new_item_queue
1624
if item not in all_old_items]
1625
self._new_item_queue = []
1627
yield None, new_items
1628
refs = refs.difference(all_old_chks)
1629
processed_new_refs.update(refs)
1631
# TODO: Using a SimpleSet for self._processed_new_refs and
1632
# saved as much as 10MB of peak memory. However, it requires
1633
# implementing a non-pyrex version.
1635
next_refs_update = next_refs.update
1636
# Inlining _read_nodes_from_store improves 'bzr branch bzr.dev'
1637
# from 1m54s to 1m51s. Consider it.
1638
for record, _, p_refs, items in self._read_nodes_from_store(refs):
1640
# using the 'if' check saves about 145s => 141s, when
1641
# streaming initial branch of Launchpad data.
1642
items = [item for item in items
1643
if item not in all_old_items]
1645
next_refs_update([p_r[1] for p_r in p_refs])
1647
# set1.difference(set/dict) walks all of set1, and checks if it
1648
# exists in 'other'.
1649
# set1.difference(iterable) walks all of iterable, and does a
1650
# 'difference_update' on a clone of set1. Pick wisely based on the
1651
# expected sizes of objects.
1652
# in our case it is expected that 'new_refs' will always be quite
1654
next_refs = next_refs.difference(all_old_chks)
1655
next_refs = next_refs.difference(processed_new_refs)
1656
processed_new_refs.update(next_refs)
1659
def _process_next_old(self):
1660
# Since we don't filter uninteresting any further than during
1661
# _read_all_roots, process the whole queue in a single pass.
1662
refs = self._old_queue
1663
self._old_queue = []
1664
all_old_chks = self._all_old_chks
1665
for record, _, prefix_refs, items in self._read_nodes_from_store(refs):
1666
# TODO: Use StaticTuple here?
1667
self._all_old_items.update(items)
1668
refs = [r for _,r in prefix_refs if r not in all_old_chks]
1669
self._old_queue.extend(refs)
1670
all_old_chks.update(refs)
1672
def _process_queues(self):
1673
while self._old_queue:
1674
self._process_next_old()
1675
return self._flush_new_queue()
1678
for record in self._read_all_roots():
1680
for record, items in self._process_queues():
1684
def iter_interesting_nodes(store, interesting_root_keys,
1685
uninteresting_root_keys, pb=None):
1686
"""Given root keys, find interesting nodes.
1688
Evaluate nodes referenced by interesting_root_keys. Ones that are also
1689
referenced from uninteresting_root_keys are not considered interesting.
1691
:param interesting_root_keys: keys which should be part of the
1692
"interesting" nodes (which will be yielded)
1693
:param uninteresting_root_keys: keys which should be filtered out of the
1696
(interesting record, {interesting key:values})
1698
iterator = CHKMapDifference(store, interesting_root_keys,
1699
uninteresting_root_keys,
1700
search_key_func=store._search_key_func,
1702
return iterator.process()
1706
from bzrlib._chk_map_pyx import (
1709
_deserialise_leaf_node,
1710
_deserialise_internal_node,
1712
except ImportError, e:
1713
osutils.failed_to_load_extension(e)
1714
from bzrlib._chk_map_py import (
1717
_deserialise_leaf_node,
1718
_deserialise_internal_node,
1720
search_key_registry.register('hash-16-way', _search_key_16)
1721
search_key_registry.register('hash-255-way', _search_key_255)
1724
def _check_key(key):
1725
"""Helper function to assert that a key is properly formatted.
1727
This generally shouldn't be used in production code, but it can be helpful
1730
if type(key) is not StaticTuple:
1731
raise TypeError('key %r is not StaticTuple but %s' % (key, type(key)))
1733
raise ValueError('key %r should have length 1, not %d' % (key, len(key),))
1734
if type(key[0]) is not str:
1735
raise TypeError('key %r should hold a str, not %r'
1736
% (key, type(key[0])))
1737
if not key[0].startswith('sha1:'):
1738
raise ValueError('key %r should point to a sha1:' % (key,))
added
removed
bzrlib/chk_map.py
