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(), """
56
from bzrlib.static_tuple import StaticTuple
59
# If each line is 50 bytes, and you have 255 internal pages, with 255-way fan
60
# out, it takes 3.1MB to cache the layer.
61
_PAGE_CACHE_SIZE = 4*1024*1024
62
# We are caching bytes so len(value) is perfectly accurate
63
_page_cache = lru_cache.LRUSizeCache(_PAGE_CACHE_SIZE)
68
# If a ChildNode falls below this many bytes, we check for a remap
69
_INTERESTING_NEW_SIZE = 50
70
# If a ChildNode shrinks by more than this amount, we check for a remap
71
_INTERESTING_SHRINKAGE_LIMIT = 20
72
# If we delete more than this many nodes applying a delta, we check for a remap
73
_INTERESTING_DELETES_LIMIT = 5
76
def _search_key_plain(key):
77
"""Map the key tuple into a search string that just uses the key bytes."""
78
return '\x00'.join(key)
81
search_key_registry = registry.Registry()
82
search_key_registry.register('plain', _search_key_plain)
86
"""A persistent map from string to string backed by a CHK store."""
88
__slots__ = ('_store', '_root_node', '_search_key_func')
90
def __init__(self, store, root_key, search_key_func=None):
91
"""Create a CHKMap object.
93
:param store: The store the CHKMap is stored in.
94
:param root_key: The root key of the map. None to create an empty
96
:param search_key_func: A function mapping a key => bytes. These bytes
97
are then used by the internal nodes to split up leaf nodes into
101
if search_key_func is None:
102
search_key_func = _search_key_plain
103
self._search_key_func = search_key_func
105
self._root_node = LeafNode(search_key_func=search_key_func)
107
self._root_node = self._node_key(root_key)
109
def apply_delta(self, delta):
110
"""Apply a delta to the map.
112
:param delta: An iterable of old_key, new_key, new_value tuples.
113
If new_key is not None, then new_key->new_value is inserted
114
into the map; if old_key is not None, then the old mapping
115
of old_key is removed.
118
# Check preconditions first.
119
as_st = StaticTuple.from_sequence
120
new_items = set([as_st(key) for (old, key, value) in delta
121
if key is not None and old is None])
122
existing_new = list(self.iteritems(key_filter=new_items))
124
raise errors.InconsistentDeltaDelta(delta,
125
"New items are already in the map %r." % existing_new)
127
for old, new, value in delta:
128
if old is not None and old != new:
129
self.unmap(old, check_remap=False)
131
for old, new, value in delta:
134
if delete_count > _INTERESTING_DELETES_LIMIT:
135
trace.mutter("checking remap as %d deletions", delete_count)
139
def _ensure_root(self):
140
"""Ensure that the root node is an object not a key."""
141
if type(self._root_node) is StaticTuple:
142
# Demand-load the root
143
self._root_node = self._get_node(self._root_node)
145
def _get_node(self, node):
148
Note that this does not update the _items dict in objects containing a
149
reference to this node. As such it does not prevent subsequent IO being
152
:param node: A tuple key or node object.
153
:return: A node object.
155
if type(node) is StaticTuple:
156
bytes = self._read_bytes(node)
157
return _deserialise(bytes, node,
158
search_key_func=self._search_key_func)
162
def _read_bytes(self, key):
164
return _page_cache[key]
166
stream = self._store.get_record_stream([key], 'unordered', True)
167
bytes = stream.next().get_bytes_as('fulltext')
168
_page_cache[key] = bytes
171
def _dump_tree(self, include_keys=False):
172
"""Return the tree in a string representation."""
174
res = self._dump_tree_node(self._root_node, prefix='', indent='',
175
include_keys=include_keys)
176
res.append('') # Give a trailing '\n'
177
return '\n'.join(res)
179
def _dump_tree_node(self, node, prefix, indent, include_keys=True):
180
"""For this node and all children, generate a string representation."""
185
node_key = node.key()
186
if node_key is not None:
187
key_str = ' %s' % (node_key[0],)
190
result.append('%s%r %s%s' % (indent, prefix, node.__class__.__name__,
192
if type(node) is InternalNode:
193
# Trigger all child nodes to get loaded
194
list(node._iter_nodes(self._store))
195
for prefix, sub in sorted(node._items.iteritems()):
196
result.extend(self._dump_tree_node(sub, prefix, indent + ' ',
197
include_keys=include_keys))
199
for key, value in sorted(node._items.iteritems()):
200
# Don't use prefix nor indent here to line up when used in
201
# tests in conjunction with assertEqualDiff
202
result.append(' %r %r' % (tuple(key), value))
206
def from_dict(klass, store, initial_value, maximum_size=0, key_width=1,
207
search_key_func=None):
208
"""Create a CHKMap in store with initial_value as the content.
210
:param store: The store to record initial_value in, a VersionedFiles
211
object with 1-tuple keys supporting CHK key generation.
212
:param initial_value: A dict to store in store. Its keys and values
214
:param maximum_size: The maximum_size rule to apply to nodes. This
215
determines the size at which no new data is added to a single node.
216
:param key_width: The number of elements in each key_tuple being stored
218
:param search_key_func: A function mapping a key => bytes. These bytes
219
are then used by the internal nodes to split up leaf nodes into
221
:return: The root chk of the resulting CHKMap.
223
root_key = klass._create_directly(store, initial_value,
224
maximum_size=maximum_size, key_width=key_width,
225
search_key_func=search_key_func)
226
if type(root_key) is not StaticTuple:
227
raise AssertionError('we got a %s instead of a StaticTuple'
232
def _create_via_map(klass, store, initial_value, maximum_size=0,
233
key_width=1, search_key_func=None):
234
result = klass(store, None, search_key_func=search_key_func)
235
result._root_node.set_maximum_size(maximum_size)
236
result._root_node._key_width = key_width
238
for key, value in initial_value.items():
239
delta.append((None, key, value))
240
root_key = result.apply_delta(delta)
244
def _create_directly(klass, store, initial_value, maximum_size=0,
245
key_width=1, search_key_func=None):
246
node = LeafNode(search_key_func=search_key_func)
247
node.set_maximum_size(maximum_size)
248
node._key_width = key_width
249
as_st = StaticTuple.from_sequence
250
node._items = dict([(as_st(key), val) for key, val
251
in initial_value.iteritems()])
252
node._raw_size = sum([node._key_value_len(key, value)
253
for key,value in node._items.iteritems()])
254
node._len = len(node._items)
255
node._compute_search_prefix()
256
node._compute_serialised_prefix()
259
and node._current_size() > maximum_size):
260
prefix, node_details = node._split(store)
261
if len(node_details) == 1:
262
raise AssertionError('Failed to split using node._split')
263
node = InternalNode(prefix, search_key_func=search_key_func)
264
node.set_maximum_size(maximum_size)
265
node._key_width = key_width
266
for split, subnode in node_details:
267
node.add_node(split, subnode)
268
keys = list(node.serialise(store))
271
def iter_changes(self, basis):
272
"""Iterate over the changes between basis and self.
274
:return: An iterator of tuples: (key, old_value, new_value). Old_value
275
is None for keys only in self; new_value is None for keys only in
279
# Read both trees in lexographic, highest-first order.
280
# Any identical nodes we skip
281
# Any unique prefixes we output immediately.
282
# values in a leaf node are treated as single-value nodes in the tree
283
# which allows them to be not-special-cased. We know to output them
284
# because their value is a string, not a key(tuple) or node.
286
# corner cases to beware of when considering this function:
287
# *) common references are at different heights.
288
# consider two trees:
289
# {'a': LeafNode={'aaa':'foo', 'aab':'bar'}, 'b': LeafNode={'b'}}
290
# {'a': InternalNode={'aa':LeafNode={'aaa':'foo', 'aab':'bar'},
291
# 'ab':LeafNode={'ab':'bar'}}
292
# 'b': LeafNode={'b'}}
293
# the node with aaa/aab will only be encountered in the second tree
294
# after reading the 'a' subtree, but it is encountered in the first
295
# tree immediately. Variations on this may have read internal nodes
296
# like this. we want to cut the entire pending subtree when we
297
# realise we have a common node. For this we use a list of keys -
298
# the path to a node - and check the entire path is clean as we
300
if self._node_key(self._root_node) == self._node_key(basis._root_node):
304
excluded_keys = set()
305
self_node = self._root_node
306
basis_node = basis._root_node
307
# A heap, each element is prefix, node(tuple/NodeObject/string),
308
# key_path (a list of tuples, tail-sharing down the tree.)
311
def process_node(node, path, a_map, pending):
312
# take a node and expand it
313
node = a_map._get_node(node)
314
if type(node) == LeafNode:
315
path = (node._key, path)
316
for key, value in node._items.items():
317
# For a LeafNode, the key is a serialized_key, rather than
318
# a search_key, but the heap is using search_keys
319
search_key = node._search_key_func(key)
320
heapq.heappush(pending, (search_key, key, value, path))
322
# type(node) == InternalNode
323
path = (node._key, path)
324
for prefix, child in node._items.items():
325
heapq.heappush(pending, (prefix, None, child, path))
326
def process_common_internal_nodes(self_node, basis_node):
327
self_items = set(self_node._items.items())
328
basis_items = set(basis_node._items.items())
329
path = (self_node._key, None)
330
for prefix, child in self_items - basis_items:
331
heapq.heappush(self_pending, (prefix, None, child, path))
332
path = (basis_node._key, None)
333
for prefix, child in basis_items - self_items:
334
heapq.heappush(basis_pending, (prefix, None, child, path))
335
def process_common_leaf_nodes(self_node, basis_node):
336
self_items = set(self_node._items.items())
337
basis_items = set(basis_node._items.items())
338
path = (self_node._key, None)
339
for key, value in self_items - basis_items:
340
prefix = self._search_key_func(key)
341
heapq.heappush(self_pending, (prefix, key, value, path))
342
path = (basis_node._key, None)
343
for key, value in basis_items - self_items:
344
prefix = basis._search_key_func(key)
345
heapq.heappush(basis_pending, (prefix, key, value, path))
346
def process_common_prefix_nodes(self_node, self_path,
347
basis_node, basis_path):
348
# Would it be more efficient if we could request both at the same
350
self_node = self._get_node(self_node)
351
basis_node = basis._get_node(basis_node)
352
if (type(self_node) == InternalNode
353
and type(basis_node) == InternalNode):
354
# Matching internal nodes
355
process_common_internal_nodes(self_node, basis_node)
356
elif (type(self_node) == LeafNode
357
and type(basis_node) == LeafNode):
358
process_common_leaf_nodes(self_node, basis_node)
360
process_node(self_node, self_path, self, self_pending)
361
process_node(basis_node, basis_path, basis, basis_pending)
362
process_common_prefix_nodes(self_node, None, basis_node, None)
365
excluded_keys = set()
366
def check_excluded(key_path):
367
# Note that this is N^2, it depends on us trimming trees
368
# aggressively to not become slow.
369
# A better implementation would probably have a reverse map
370
# back to the children of a node, and jump straight to it when
371
# a common node is detected, the proceed to remove the already
372
# pending children. bzrlib.graph has a searcher module with a
374
while key_path is not None:
375
key, key_path = key_path
376
if key in excluded_keys:
381
while self_pending or basis_pending:
384
# self is exhausted: output remainder of basis
385
for prefix, key, node, path in basis_pending:
386
if check_excluded(path):
388
node = basis._get_node(node)
391
yield (key, node, None)
393
# subtree - fastpath the entire thing.
394
for key, value in node.iteritems(basis._store):
395
yield (key, value, None)
397
elif not basis_pending:
398
# basis is exhausted: output remainder of self.
399
for prefix, key, node, path in self_pending:
400
if check_excluded(path):
402
node = self._get_node(node)
405
yield (key, None, node)
407
# subtree - fastpath the entire thing.
408
for key, value in node.iteritems(self._store):
409
yield (key, None, value)
412
# XXX: future optimisation - yield the smaller items
413
# immediately rather than pushing everything on/off the
414
# heaps. Applies to both internal nodes and leafnodes.
415
if self_pending[0][0] < basis_pending[0][0]:
417
prefix, key, node, path = heapq.heappop(self_pending)
418
if check_excluded(path):
422
yield (key, None, node)
424
process_node(node, path, self, self_pending)
426
elif self_pending[0][0] > basis_pending[0][0]:
428
prefix, key, node, path = heapq.heappop(basis_pending)
429
if check_excluded(path):
433
yield (key, node, None)
435
process_node(node, path, basis, basis_pending)
438
# common prefix: possibly expand both
439
if self_pending[0][1] is None:
444
if basis_pending[0][1] is None:
449
if not read_self and not read_basis:
450
# compare a common value
451
self_details = heapq.heappop(self_pending)
452
basis_details = heapq.heappop(basis_pending)
453
if self_details[2] != basis_details[2]:
454
yield (self_details[1],
455
basis_details[2], self_details[2])
457
# At least one side wasn't a simple value
458
if (self._node_key(self_pending[0][2]) ==
459
self._node_key(basis_pending[0][2])):
460
# Identical pointers, skip (and don't bother adding to
461
# excluded, it won't turn up again.
462
heapq.heappop(self_pending)
463
heapq.heappop(basis_pending)
465
# Now we need to expand this node before we can continue
466
if read_self and read_basis:
467
# Both sides start with the same prefix, so process
469
self_prefix, _, self_node, self_path = heapq.heappop(
471
basis_prefix, _, basis_node, basis_path = heapq.heappop(
473
if self_prefix != basis_prefix:
474
raise AssertionError(
475
'%r != %r' % (self_prefix, basis_prefix))
476
process_common_prefix_nodes(
477
self_node, self_path,
478
basis_node, basis_path)
481
prefix, key, node, path = heapq.heappop(self_pending)
482
if check_excluded(path):
484
process_node(node, path, self, self_pending)
486
prefix, key, node, path = heapq.heappop(basis_pending)
487
if check_excluded(path):
489
process_node(node, path, basis, basis_pending)
492
def iteritems(self, key_filter=None):
493
"""Iterate over the entire CHKMap's contents."""
495
if key_filter is not None:
496
as_st = StaticTuple.from_sequence
497
key_filter = [as_st(key) for key in key_filter]
498
return self._root_node.iteritems(self._store, key_filter=key_filter)
501
"""Return the key for this map."""
502
if type(self._root_node) is StaticTuple:
503
return self._root_node
505
return self._root_node._key
509
return len(self._root_node)
511
def map(self, key, value):
512
"""Map a key tuple to value.
514
:param key: A key to map.
515
:param value: The value to assign to key.
517
key = StaticTuple.from_sequence(key)
518
# Need a root object.
520
prefix, node_details = self._root_node.map(self._store, key, value)
521
if len(node_details) == 1:
522
self._root_node = node_details[0][1]
524
self._root_node = InternalNode(prefix,
525
search_key_func=self._search_key_func)
526
self._root_node.set_maximum_size(node_details[0][1].maximum_size)
527
self._root_node._key_width = node_details[0][1]._key_width
528
for split, node in node_details:
529
self._root_node.add_node(split, node)
531
def _node_key(self, node):
532
"""Get the key for a node whether it's a tuple or node."""
533
if type(node) is tuple:
534
node = StaticTuple.from_sequence(node)
535
if type(node) is StaticTuple:
540
def unmap(self, key, check_remap=True):
541
"""remove key from the map."""
542
key = StaticTuple.from_sequence(key)
544
if type(self._root_node) is InternalNode:
545
unmapped = self._root_node.unmap(self._store, key,
546
check_remap=check_remap)
548
unmapped = self._root_node.unmap(self._store, key)
549
self._root_node = unmapped
551
def _check_remap(self):
552
"""Check if nodes can be collapsed."""
554
if type(self._root_node) is InternalNode:
555
self._root_node._check_remap(self._store)
558
"""Save the map completely.
560
:return: The key of the root node.
562
if type(self._root_node) is StaticTuple:
564
return self._root_node
565
keys = list(self._root_node.serialise(self._store))
570
"""Base class defining the protocol for CHK Map nodes.
572
:ivar _raw_size: The total size of the serialized key:value data, before
573
adding the header bytes, and without prefix compression.
576
__slots__ = ('_key', '_len', '_maximum_size', '_key_width',
577
'_raw_size', '_items', '_search_prefix', '_search_key_func'
580
def __init__(self, key_width=1):
583
:param key_width: The width of keys for this node.
586
# Current number of elements
588
self._maximum_size = 0
589
self._key_width = key_width
590
# current size in bytes
592
# The pointers/values this node has - meaning defined by child classes.
594
# The common search prefix
595
self._search_prefix = None
598
items_str = str(sorted(self._items))
599
if len(items_str) > 20:
600
items_str = items_str[:16] + '...]'
601
return '%s(key:%s len:%s size:%s max:%s prefix:%s items:%s)' % (
602
self.__class__.__name__, self._key, self._len, self._raw_size,
603
self._maximum_size, self._search_prefix, items_str)
612
def maximum_size(self):
613
"""What is the upper limit for adding references to a node."""
614
return self._maximum_size
616
def set_maximum_size(self, new_size):
617
"""Set the size threshold for nodes.
619
:param new_size: The size at which no data is added to a node. 0 for
622
self._maximum_size = new_size
625
def common_prefix(cls, prefix, key):
626
"""Given 2 strings, return the longest prefix common to both.
628
:param prefix: This has been the common prefix for other keys, so it is
629
more likely to be the common prefix in this case as well.
630
:param key: Another string to compare to
632
if key.startswith(prefix):
635
# Is there a better way to do this?
636
for pos, (left, right) in enumerate(zip(prefix, key)):
640
common = prefix[:pos+1]
644
def common_prefix_for_keys(cls, keys):
645
"""Given a list of keys, find their common prefix.
647
:param keys: An iterable of strings.
648
:return: The longest common prefix of all keys.
652
if common_prefix is None:
655
common_prefix = cls.common_prefix(common_prefix, key)
656
if not common_prefix:
657
# if common_prefix is the empty string, then we know it won't
663
# Singleton indicating we have not computed _search_prefix yet
666
class LeafNode(Node):
667
"""A node containing actual key:value pairs.
669
:ivar _items: A dict of key->value items. The key is in tuple form.
670
:ivar _size: The number of bytes that would be used by serializing all of
674
__slots__ = ('_common_serialised_prefix', '_serialise_key')
676
def __init__(self, search_key_func=None):
678
# All of the keys in this leaf node share this common prefix
679
self._common_serialised_prefix = None
680
self._serialise_key = '\x00'.join
681
if search_key_func is None:
682
self._search_key_func = _search_key_plain
684
self._search_key_func = search_key_func
687
items_str = str(sorted(self._items))
688
if len(items_str) > 20:
689
items_str = items_str[:16] + '...]'
691
'%s(key:%s len:%s size:%s max:%s prefix:%s keywidth:%s items:%s)' \
692
% (self.__class__.__name__, self._key, self._len, self._raw_size,
693
self._maximum_size, self._search_prefix, self._key_width, items_str)
695
def _current_size(self):
696
"""Answer the current serialised size of this node.
698
This differs from self._raw_size in that it includes the bytes used for
701
if self._common_serialised_prefix is None:
705
# We will store a single string with the common prefix
706
# And then that common prefix will not be stored in any of the
708
prefix_len = len(self._common_serialised_prefix)
709
bytes_for_items = (self._raw_size - (prefix_len * self._len))
710
return (9 # 'chkleaf:\n'
711
+ len(str(self._maximum_size)) + 1
712
+ len(str(self._key_width)) + 1
713
+ len(str(self._len)) + 1
718
def deserialise(klass, bytes, key, search_key_func=None):
719
"""Deserialise bytes, with key key, into a LeafNode.
721
:param bytes: The bytes of the node.
722
:param key: The key that the serialised node has.
724
key = static_tuple.expect_static_tuple(key)
725
return _deserialise_leaf_node(bytes, key,
726
search_key_func=search_key_func)
728
def iteritems(self, store, key_filter=None):
729
"""Iterate over items in the node.
731
:param key_filter: A filter to apply to the node. It should be a
732
list/set/dict or similar repeatedly iterable container.
734
if key_filter is not None:
735
# Adjust the filter - short elements go to a prefix filter. All
736
# other items are looked up directly.
737
# XXX: perhaps defaultdict? Profiling<rinse and repeat>
739
for key in key_filter:
740
if len(key) == self._key_width:
741
# This filter is meant to match exactly one key, yield it
744
yield key, self._items[key]
746
# This key is not present in this map, continue
749
# Short items, we need to match based on a prefix
750
length_filter = filters.setdefault(len(key), set())
751
length_filter.add(key)
753
filters = filters.items()
754
for item in self._items.iteritems():
755
for length, length_filter in filters:
756
if item[0][:length] in length_filter:
760
for item in self._items.iteritems():
763
def _key_value_len(self, key, value):
764
# TODO: Should probably be done without actually joining the key, but
765
# then that can be done via the C extension
766
return (len(self._serialise_key(key)) + 1
767
+ len(str(value.count('\n'))) + 1
770
def _search_key(self, key):
771
return self._search_key_func(key)
773
def _map_no_split(self, key, value):
774
"""Map a key to a value.
776
This assumes either the key does not already exist, or you have already
777
removed its size and length from self.
779
:return: True if adding this node should cause us to split.
781
self._items[key] = value
782
self._raw_size += self._key_value_len(key, value)
784
serialised_key = self._serialise_key(key)
785
if self._common_serialised_prefix is None:
786
self._common_serialised_prefix = serialised_key
788
self._common_serialised_prefix = self.common_prefix(
789
self._common_serialised_prefix, serialised_key)
790
search_key = self._search_key(key)
791
if self._search_prefix is _unknown:
792
self._compute_search_prefix()
793
if self._search_prefix is None:
794
self._search_prefix = search_key
796
self._search_prefix = self.common_prefix(
797
self._search_prefix, search_key)
799
and self._maximum_size
800
and self._current_size() > self._maximum_size):
801
# Check to see if all of the search_keys for this node are
802
# identical. We allow the node to grow under that circumstance
803
# (we could track this as common state, but it is infrequent)
804
if (search_key != self._search_prefix
805
or not self._are_search_keys_identical()):
809
def _split(self, store):
810
"""We have overflowed.
812
Split this node into multiple LeafNodes, return it up the stack so that
813
the next layer creates a new InternalNode and references the new nodes.
815
:return: (common_serialised_prefix, [(node_serialised_prefix, node)])
817
if self._search_prefix is _unknown:
818
raise AssertionError('Search prefix must be known')
819
common_prefix = self._search_prefix
820
split_at = len(common_prefix) + 1
822
for key, value in self._items.iteritems():
823
search_key = self._search_key(key)
824
prefix = search_key[:split_at]
825
# TODO: Generally only 1 key can be exactly the right length,
826
# which means we can only have 1 key in the node pointed
827
# at by the 'prefix\0' key. We might want to consider
828
# folding it into the containing InternalNode rather than
829
# having a fixed length-1 node.
830
# Note this is probably not true for hash keys, as they
831
# may get a '\00' node anywhere, but won't have keys of
833
if len(prefix) < split_at:
834
prefix += '\x00'*(split_at - len(prefix))
835
if prefix not in result:
836
node = LeafNode(search_key_func=self._search_key_func)
837
node.set_maximum_size(self._maximum_size)
838
node._key_width = self._key_width
839
result[prefix] = node
841
node = result[prefix]
842
sub_prefix, node_details = node.map(store, key, value)
843
if len(node_details) > 1:
844
if prefix != sub_prefix:
845
# This node has been split and is now found via a different
848
new_node = InternalNode(sub_prefix,
849
search_key_func=self._search_key_func)
850
new_node.set_maximum_size(self._maximum_size)
851
new_node._key_width = self._key_width
852
for split, node in node_details:
853
new_node.add_node(split, node)
854
result[prefix] = new_node
855
return common_prefix, result.items()
857
def map(self, store, key, value):
858
"""Map key to value."""
859
if key in self._items:
860
self._raw_size -= self._key_value_len(key, self._items[key])
863
if self._map_no_split(key, value):
864
return self._split(store)
866
if self._search_prefix is _unknown:
867
raise AssertionError('%r must be known' % self._search_prefix)
868
return self._search_prefix, [("", self)]
870
def serialise(self, store):
871
"""Serialise the LeafNode to store.
873
:param store: A VersionedFiles honouring the CHK extensions.
874
:return: An iterable of the keys inserted by this operation.
876
lines = ["chkleaf:\n"]
877
lines.append("%d\n" % self._maximum_size)
878
lines.append("%d\n" % self._key_width)
879
lines.append("%d\n" % self._len)
880
if self._common_serialised_prefix is None:
882
if len(self._items) != 0:
883
raise AssertionError('If _common_serialised_prefix is None'
884
' we should have no items')
886
lines.append('%s\n' % (self._common_serialised_prefix,))
887
prefix_len = len(self._common_serialised_prefix)
888
for key, value in sorted(self._items.items()):
889
# Always add a final newline
890
value_lines = osutils.chunks_to_lines([value + '\n'])
891
serialized = "%s\x00%s\n" % (self._serialise_key(key),
893
if not serialized.startswith(self._common_serialised_prefix):
894
raise AssertionError('We thought the common prefix was %r'
895
' but entry %r does not have it in common'
896
% (self._common_serialised_prefix, serialized))
897
lines.append(serialized[prefix_len:])
898
lines.extend(value_lines)
899
sha1, _, _ = store.add_lines((None,), (), lines)
900
self._key = StaticTuple("sha1:" + sha1,).intern()
901
bytes = ''.join(lines)
902
if len(bytes) != self._current_size():
903
raise AssertionError('Invalid _current_size')
904
_page_cache.add(self._key, bytes)
908
"""Return the references to other CHK's held by this node."""
911
def _compute_search_prefix(self):
912
"""Determine the common search prefix for all keys in this node.
914
:return: A bytestring of the longest search key prefix that is
915
unique within this node.
917
search_keys = [self._search_key_func(key) for key in self._items]
918
self._search_prefix = self.common_prefix_for_keys(search_keys)
919
return self._search_prefix
921
def _are_search_keys_identical(self):
922
"""Check to see if the search keys for all entries are the same.
924
When using a hash as the search_key it is possible for non-identical
925
keys to collide. If that happens enough, we may try overflow a
926
LeafNode, but as all are collisions, we must not split.
928
common_search_key = None
929
for key in self._items:
930
search_key = self._search_key(key)
931
if common_search_key is None:
932
common_search_key = search_key
933
elif search_key != common_search_key:
937
def _compute_serialised_prefix(self):
938
"""Determine the common prefix for serialised keys in this node.
940
:return: A bytestring of the longest serialised key prefix that is
941
unique within this node.
943
serialised_keys = [self._serialise_key(key) for key in self._items]
944
self._common_serialised_prefix = self.common_prefix_for_keys(
946
return self._common_serialised_prefix
948
def unmap(self, store, key):
949
"""Unmap key from the node."""
951
self._raw_size -= self._key_value_len(key, self._items[key])
953
trace.mutter("key %s not found in %r", key, self._items)
958
# Recompute from scratch
959
self._compute_search_prefix()
960
self._compute_serialised_prefix()
964
class InternalNode(Node):
965
"""A node that contains references to other nodes.
967
An InternalNode is responsible for mapping search key prefixes to child
970
:ivar _items: serialised_key => node dictionary. node may be a tuple,
971
LeafNode or InternalNode.
974
__slots__ = ('_node_width',)
976
def __init__(self, prefix='', search_key_func=None):
978
# The size of an internalnode with default values and no children.
979
# How many octets key prefixes within this node are.
981
self._search_prefix = prefix
982
if search_key_func is None:
983
self._search_key_func = _search_key_plain
985
self._search_key_func = search_key_func
987
def add_node(self, prefix, node):
988
"""Add a child node with prefix prefix, and node node.
990
:param prefix: The search key prefix for node.
991
:param node: The node being added.
993
if self._search_prefix is None:
994
raise AssertionError("_search_prefix should not be None")
995
if not prefix.startswith(self._search_prefix):
996
raise AssertionError("prefixes mismatch: %s must start with %s"
997
% (prefix,self._search_prefix))
998
if len(prefix) != len(self._search_prefix) + 1:
999
raise AssertionError("prefix wrong length: len(%s) is not %d" %
1000
(prefix, len(self._search_prefix) + 1))
1001
self._len += len(node)
1002
if not len(self._items):
1003
self._node_width = len(prefix)
1004
if self._node_width != len(self._search_prefix) + 1:
1005
raise AssertionError("node width mismatch: %d is not %d" %
1006
(self._node_width, len(self._search_prefix) + 1))
1007
self._items[prefix] = node
1010
def _current_size(self):
1011
"""Answer the current serialised size of this node."""
1012
return (self._raw_size + len(str(self._len)) + len(str(self._key_width)) +
1013
len(str(self._maximum_size)))
1016
def deserialise(klass, bytes, key, search_key_func=None):
1017
"""Deserialise bytes to an InternalNode, with key key.
1019
:param bytes: The bytes of the node.
1020
:param key: The key that the serialised node has.
1021
:return: An InternalNode instance.
1023
key = static_tuple.expect_static_tuple(key)
1024
return _deserialise_internal_node(bytes, key,
1025
search_key_func=search_key_func)
1027
def iteritems(self, store, key_filter=None):
1028
for node, node_filter in self._iter_nodes(store, key_filter=key_filter):
1029
for item in node.iteritems(store, key_filter=node_filter):
1032
def _iter_nodes(self, store, key_filter=None, batch_size=None):
1033
"""Iterate over node objects which match key_filter.
1035
:param store: A store to use for accessing content.
1036
:param key_filter: A key filter to filter nodes. Only nodes that might
1037
contain a key in key_filter will be returned.
1038
:param batch_size: If not None, then we will return the nodes that had
1039
to be read using get_record_stream in batches, rather than reading
1041
:return: An iterable of nodes. This function does not have to be fully
1042
consumed. (There will be no pending I/O when items are being returned.)
1044
# Map from chk key ('sha1:...',) to (prefix, key_filter)
1045
# prefix is the key in self._items to use, key_filter is the key_filter
1046
# entries that would match this node
1049
if key_filter is None:
1050
# yielding all nodes, yield whatever we have, and queue up a read
1051
# for whatever we are missing
1053
for prefix, node in self._items.iteritems():
1054
if node.__class__ is StaticTuple:
1055
keys[node] = (prefix, None)
1058
elif len(key_filter) == 1:
1059
# Technically, this path could also be handled by the first check
1060
# in 'self._node_width' in length_filters. However, we can handle
1061
# this case without spending any time building up the
1062
# prefix_to_keys, etc state.
1064
# This is a bit ugly, but TIMEIT showed it to be by far the fastest
1065
# 0.626us list(key_filter)[0]
1066
# is a func() for list(), 2 mallocs, and a getitem
1067
# 0.489us [k for k in key_filter][0]
1068
# still has the mallocs, avoids the func() call
1069
# 0.350us iter(key_filter).next()
1070
# has a func() call, and mallocs an iterator
1071
# 0.125us for key in key_filter: pass
1072
# no func() overhead, might malloc an iterator
1073
# 0.105us for key in key_filter: break
1074
# no func() overhead, might malloc an iterator, probably
1075
# avoids checking an 'else' clause as part of the for
1076
for key in key_filter:
1078
search_prefix = self._search_prefix_filter(key)
1079
if len(search_prefix) == self._node_width:
1080
# This item will match exactly, so just do a dict lookup, and
1081
# see what we can return
1084
node = self._items[search_prefix]
1086
# A given key can only match 1 child node, if it isn't
1087
# there, then we can just return nothing
1089
if node.__class__ is StaticTuple:
1090
keys[node] = (search_prefix, [key])
1092
# This is loaded, and the only thing that can match,
1097
# First, convert all keys into a list of search prefixes
1098
# Aggregate common prefixes, and track the keys they come from
1101
for key in key_filter:
1102
search_prefix = self._search_prefix_filter(key)
1103
length_filter = length_filters.setdefault(
1104
len(search_prefix), set())
1105
length_filter.add(search_prefix)
1106
prefix_to_keys.setdefault(search_prefix, []).append(key)
1108
if (self._node_width in length_filters
1109
and len(length_filters) == 1):
1110
# all of the search prefixes match exactly _node_width. This
1111
# means that everything is an exact match, and we can do a
1112
# lookup into self._items, rather than iterating over the items
1114
search_prefixes = length_filters[self._node_width]
1115
for search_prefix in search_prefixes:
1117
node = self._items[search_prefix]
1119
# We can ignore this one
1121
node_key_filter = prefix_to_keys[search_prefix]
1122
if node.__class__ is StaticTuple:
1123
keys[node] = (search_prefix, node_key_filter)
1125
yield node, node_key_filter
1127
# The slow way. We walk every item in self._items, and check to
1128
# see if there are any matches
1129
length_filters = length_filters.items()
1130
for prefix, node in self._items.iteritems():
1131
node_key_filter = []
1132
for length, length_filter in length_filters:
1133
sub_prefix = prefix[:length]
1134
if sub_prefix in length_filter:
1135
node_key_filter.extend(prefix_to_keys[sub_prefix])
1136
if node_key_filter: # this key matched something, yield it
1137
if node.__class__ is StaticTuple:
1138
keys[node] = (prefix, node_key_filter)
1140
yield node, node_key_filter
1142
# Look in the page cache for some more bytes
1146
bytes = _page_cache[key]
1150
node = _deserialise(bytes, key,
1151
search_key_func=self._search_key_func)
1152
prefix, node_key_filter = keys[key]
1153
self._items[prefix] = node
1155
yield node, node_key_filter
1156
for key in found_keys:
1159
# demand load some pages.
1160
if batch_size is None:
1161
# Read all the keys in
1162
batch_size = len(keys)
1163
key_order = list(keys)
1164
for batch_start in range(0, len(key_order), batch_size):
1165
batch = key_order[batch_start:batch_start + batch_size]
1166
# We have to fully consume the stream so there is no pending
1167
# I/O, so we buffer the nodes for now.
1168
stream = store.get_record_stream(batch, 'unordered', True)
1169
node_and_filters = []
1170
for record in stream:
1171
bytes = record.get_bytes_as('fulltext')
1172
node = _deserialise(bytes, record.key,
1173
search_key_func=self._search_key_func)
1174
prefix, node_key_filter = keys[record.key]
1175
node_and_filters.append((node, node_key_filter))
1176
self._items[prefix] = node
1177
_page_cache.add(record.key, bytes)
1178
for info in node_and_filters:
1181
def map(self, store, key, value):
1182
"""Map key to value."""
1183
if not len(self._items):
1184
raise AssertionError("can't map in an empty InternalNode.")
1185
search_key = self._search_key(key)
1186
if self._node_width != len(self._search_prefix) + 1:
1187
raise AssertionError("node width mismatch: %d is not %d" %
1188
(self._node_width, len(self._search_prefix) + 1))
1189
if not search_key.startswith(self._search_prefix):
1190
# This key doesn't fit in this index, so we need to split at the
1191
# point where it would fit, insert self into that internal node,
1192
# and then map this key into that node.
1193
new_prefix = self.common_prefix(self._search_prefix,
1195
new_parent = InternalNode(new_prefix,
1196
search_key_func=self._search_key_func)
1197
new_parent.set_maximum_size(self._maximum_size)
1198
new_parent._key_width = self._key_width
1199
new_parent.add_node(self._search_prefix[:len(new_prefix)+1],
1201
return new_parent.map(store, key, value)
1202
children = [node for node, _
1203
in self._iter_nodes(store, key_filter=[key])]
1208
child = self._new_child(search_key, LeafNode)
1209
old_len = len(child)
1210
if type(child) is LeafNode:
1211
old_size = child._current_size()
1214
prefix, node_details = child.map(store, key, value)
1215
if len(node_details) == 1:
1216
# child may have shrunk, or might be a new node
1217
child = node_details[0][1]
1218
self._len = self._len - old_len + len(child)
1219
self._items[search_key] = child
1222
if type(child) is LeafNode:
1223
if old_size is None:
1224
# The old node was an InternalNode which means it has now
1225
# collapsed, so we need to check if it will chain to a
1226
# collapse at this level.
1227
trace.mutter("checking remap as InternalNode -> LeafNode")
1228
new_node = self._check_remap(store)
1230
# If the LeafNode has shrunk in size, we may want to run
1231
# a remap check. Checking for a remap is expensive though
1232
# and the frequency of a successful remap is very low.
1233
# Shrinkage by small amounts is common, so we only do the
1234
# remap check if the new_size is low or the shrinkage
1235
# amount is over a configurable limit.
1236
new_size = child._current_size()
1237
shrinkage = old_size - new_size
1238
if (shrinkage > 0 and new_size < _INTERESTING_NEW_SIZE
1239
or shrinkage > _INTERESTING_SHRINKAGE_LIMIT):
1241
"checking remap as size shrunk by %d to be %d",
1242
shrinkage, new_size)
1243
new_node = self._check_remap(store)
1244
if new_node._search_prefix is None:
1245
raise AssertionError("_search_prefix should not be None")
1246
return new_node._search_prefix, [('', new_node)]
1247
# child has overflown - create a new intermediate node.
1248
# XXX: This is where we might want to try and expand our depth
1249
# to refer to more bytes of every child (which would give us
1250
# multiple pointers to child nodes, but less intermediate nodes)
1251
child = self._new_child(search_key, InternalNode)
1252
child._search_prefix = prefix
1253
for split, node in node_details:
1254
child.add_node(split, node)
1255
self._len = self._len - old_len + len(child)
1257
return self._search_prefix, [("", self)]
1259
def _new_child(self, search_key, klass):
1260
"""Create a new child node of type klass."""
1262
child.set_maximum_size(self._maximum_size)
1263
child._key_width = self._key_width
1264
child._search_key_func = self._search_key_func
1265
self._items[search_key] = child
1268
def serialise(self, store):
1269
"""Serialise the node to store.
1271
:param store: A VersionedFiles honouring the CHK extensions.
1272
:return: An iterable of the keys inserted by this operation.
1274
for node in self._items.itervalues():
1275
if type(node) is StaticTuple:
1276
# Never deserialised.
1278
if node._key is not None:
1281
for key in node.serialise(store):
1283
lines = ["chknode:\n"]
1284
lines.append("%d\n" % self._maximum_size)
1285
lines.append("%d\n" % self._key_width)
1286
lines.append("%d\n" % self._len)
1287
if self._search_prefix is None:
1288
raise AssertionError("_search_prefix should not be None")
1289
lines.append('%s\n' % (self._search_prefix,))
1290
prefix_len = len(self._search_prefix)
1291
for prefix, node in sorted(self._items.items()):
1292
if type(node) is StaticTuple:
1296
serialised = "%s\x00%s\n" % (prefix, key)
1297
if not serialised.startswith(self._search_prefix):
1298
raise AssertionError("prefixes mismatch: %s must start with %s"
1299
% (serialised, self._search_prefix))
1300
lines.append(serialised[prefix_len:])
1301
sha1, _, _ = store.add_lines((None,), (), lines)
1302
self._key = StaticTuple("sha1:" + sha1,).intern()
1303
_page_cache.add(self._key, ''.join(lines))
1306
def _search_key(self, key):
1307
"""Return the serialised key for key in this node."""
1308
# search keys are fixed width. All will be self._node_width wide, so we
1310
return (self._search_key_func(key) + '\x00'*self._node_width)[:self._node_width]
1312
def _search_prefix_filter(self, key):
1313
"""Serialise key for use as a prefix filter in iteritems."""
1314
return self._search_key_func(key)[:self._node_width]
1316
def _split(self, offset):
1317
"""Split this node into smaller nodes starting at offset.
1319
:param offset: The offset to start the new child nodes at.
1320
:return: An iterable of (prefix, node) tuples. prefix is a byte
1321
prefix for reaching node.
1323
if offset >= self._node_width:
1324
for node in self._items.values():
1325
for result in node._split(offset):
1328
for key, node in self._items.items():
1332
"""Return the references to other CHK's held by this node."""
1333
if self._key is None:
1334
raise AssertionError("unserialised nodes have no refs.")
1336
for value in self._items.itervalues():
1337
if type(value) is StaticTuple:
1340
refs.append(value.key())
1343
def _compute_search_prefix(self, extra_key=None):
1344
"""Return the unique key prefix for this node.
1346
:return: A bytestring of the longest search key prefix that is
1347
unique within this node.
1349
self._search_prefix = self.common_prefix_for_keys(self._items)
1350
return self._search_prefix
1352
def unmap(self, store, key, check_remap=True):
1353
"""Remove key from this node and it's children."""
1354
if not len(self._items):
1355
raise AssertionError("can't unmap in an empty InternalNode.")
1356
children = [node for node, _
1357
in self._iter_nodes(store, key_filter=[key])]
1363
unmapped = child.unmap(store, key)
1365
search_key = self._search_key(key)
1366
if len(unmapped) == 0:
1367
# All child nodes are gone, remove the child:
1368
del self._items[search_key]
1371
# Stash the returned node
1372
self._items[search_key] = unmapped
1373
if len(self._items) == 1:
1374
# this node is no longer needed:
1375
return self._items.values()[0]
1376
if type(unmapped) is InternalNode:
1379
return self._check_remap(store)
1383
def _check_remap(self, store):
1384
"""Check if all keys contained by children fit in a single LeafNode.
1386
:param store: A store to use for reading more nodes
1387
:return: Either self, or a new LeafNode which should replace self.
1389
# Logic for how we determine when we need to rebuild
1390
# 1) Implicitly unmap() is removing a key which means that the child
1391
# nodes are going to be shrinking by some extent.
1392
# 2) If all children are LeafNodes, it is possible that they could be
1393
# combined into a single LeafNode, which can then completely replace
1394
# this internal node with a single LeafNode
1395
# 3) If *one* child is an InternalNode, we assume it has already done
1396
# all the work to determine that its children cannot collapse, and
1397
# we can then assume that those nodes *plus* the current nodes don't
1398
# have a chance of collapsing either.
1399
# So a very cheap check is to just say if 'unmapped' is an
1400
# InternalNode, we don't have to check further.
1402
# TODO: Another alternative is to check the total size of all known
1403
# LeafNodes. If there is some formula we can use to determine the
1404
# final size without actually having to read in any more
1405
# children, it would be nice to have. However, we have to be
1406
# careful with stuff like nodes that pull out the common prefix
1407
# of each key, as adding a new key can change the common prefix
1408
# and cause size changes greater than the length of one key.
1409
# So for now, we just add everything to a new Leaf until it
1410
# splits, as we know that will give the right answer
1411
new_leaf = LeafNode(search_key_func=self._search_key_func)
1412
new_leaf.set_maximum_size(self._maximum_size)
1413
new_leaf._key_width = self._key_width
1414
# A batch_size of 16 was chosen because:
1415
# a) In testing, a 4k page held 14 times. So if we have more than 16
1416
# leaf nodes we are unlikely to hold them in a single new leaf
1417
# node. This still allows for 1 round trip
1418
# b) With 16-way fan out, we can still do a single round trip
1419
# c) With 255-way fan out, we don't want to read all 255 and destroy
1420
# the page cache, just to determine that we really don't need it.
1421
for node, _ in self._iter_nodes(store, batch_size=16):
1422
if type(node) is InternalNode:
1423
# Without looking at any leaf nodes, we are sure
1425
for key, value in node._items.iteritems():
1426
if new_leaf._map_no_split(key, value):
1428
trace.mutter("remap generated a new LeafNode")
1432
def _deserialise(bytes, key, search_key_func):
1433
"""Helper for repositorydetails - convert bytes to a node."""
1434
if bytes.startswith("chkleaf:\n"):
1435
node = LeafNode.deserialise(bytes, key, search_key_func=search_key_func)
1436
elif bytes.startswith("chknode:\n"):
1437
node = InternalNode.deserialise(bytes, key,
1438
search_key_func=search_key_func)
1440
raise AssertionError("Unknown node type.")
1444
class CHKMapDifference(object):
1445
"""Iterate the stored pages and key,value pairs for (new - old).
1447
This class provides a generator over the stored CHK pages and the
1448
(key, value) pairs that are in any of the new maps and not in any of the
1451
Note that it may yield chk pages that are common (especially root nodes),
1452
but it won't yield (key,value) pairs that are common.
1455
def __init__(self, store, new_root_keys, old_root_keys,
1456
search_key_func, pb=None):
1457
# TODO: Should we add a StaticTuple barrier here? It would be nice to
1458
# force callers to use StaticTuple, because there will often be
1459
# lots of keys passed in here. And even if we cast it locally,
1460
# that just meanst that we will have *both* a StaticTuple and a
1461
# tuple() in memory, referring to the same object. (so a net
1462
# increase in memory, not a decrease.)
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
# TODO: This might grow to a large size if there are lots of merge
1471
# parents, etc. However, it probably doesn't scale to O(history)
1472
# like _processed_new_refs does.
1473
self._all_old_chks = set(self._old_root_keys)
1474
# All items that we have seen from the old_root_keys
1475
self._all_old_items = set()
1476
# These are interesting items which were either read, or already in the
1477
# interesting queue (so we don't need to walk them again)
1478
# TODO: processed_new_refs becomes O(all_chks), consider switching to
1480
self._processed_new_refs = set()
1481
self._search_key_func = search_key_func
1483
# The uninteresting and interesting nodes to be searched
1484
self._old_queue = []
1485
self._new_queue = []
1486
# Holds the (key, value) items found when processing the root nodes,
1487
# waiting for the uninteresting nodes to be walked
1488
self._new_item_queue = []
1491
def _read_nodes_from_store(self, keys):
1492
# We chose not to use _page_cache, because we think in terms of records
1493
# to be yielded. Also, we expect to touch each page only 1 time during
1494
# this code. (We may want to evaluate saving the raw bytes into the
1495
# page cache, which would allow a working tree update after the fetch
1496
# to not have to read the bytes again.)
1497
as_st = StaticTuple.from_sequence
1498
stream = self._store.get_record_stream(keys, 'unordered', True)
1499
for record in stream:
1500
if self._pb is not None:
1502
if record.storage_kind == 'absent':
1503
raise errors.NoSuchRevision(self._store, record.key)
1504
bytes = record.get_bytes_as('fulltext')
1505
node = _deserialise(bytes, record.key,
1506
search_key_func=self._search_key_func)
1507
if type(node) is InternalNode:
1508
# Note we don't have to do node.refs() because we know that
1509
# there are no children that have been pushed into this node
1510
# Note: Using as_st() here seemed to save 1.2MB, which would
1511
# indicate that we keep 100k prefix_refs around while
1512
# processing. They *should* be shorter lived than that...
1513
# It does cost us ~10s of processing time
1514
#prefix_refs = [as_st(item) for item in node._items.iteritems()]
1515
prefix_refs = node._items.items()
1519
# Note: We don't use a StaticTuple here. Profiling showed a
1520
# minor memory improvement (0.8MB out of 335MB peak 0.2%)
1521
# But a significant slowdown (15s / 145s, or 10%)
1522
items = node._items.items()
1523
yield record, node, prefix_refs, items
1525
def _read_old_roots(self):
1526
old_chks_to_enqueue = []
1527
all_old_chks = self._all_old_chks
1528
for record, node, prefix_refs, items in \
1529
self._read_nodes_from_store(self._old_root_keys):
1530
# Uninteresting node
1531
prefix_refs = [p_r for p_r in prefix_refs
1532
if p_r[1] not in all_old_chks]
1533
new_refs = [p_r[1] for p_r in prefix_refs]
1534
all_old_chks.update(new_refs)
1535
# TODO: This might be a good time to turn items into StaticTuple
1536
# instances and possibly intern them. However, this does not
1537
# impact 'initial branch' performance, so I'm not worrying
1539
self._all_old_items.update(items)
1540
# Queue up the uninteresting references
1541
# Don't actually put them in the 'to-read' queue until we have
1542
# finished checking the interesting references
1543
old_chks_to_enqueue.extend(prefix_refs)
1544
return old_chks_to_enqueue
1546
def _enqueue_old(self, new_prefixes, old_chks_to_enqueue):
1547
# At this point, we have read all the uninteresting and interesting
1548
# items, so we can queue up the uninteresting stuff, knowing that we've
1549
# handled the interesting ones
1550
for prefix, ref in old_chks_to_enqueue:
1551
not_interesting = True
1552
for i in xrange(len(prefix), 0, -1):
1553
if prefix[:i] in new_prefixes:
1554
not_interesting = False
1557
# This prefix is not part of the remaining 'interesting set'
1559
self._old_queue.append(ref)
1561
def _read_all_roots(self):
1562
"""Read the root pages.
1564
This is structured as a generator, so that the root records can be
1565
yielded up to whoever needs them without any buffering.
1567
# This is the bootstrap phase
1568
if not self._old_root_keys:
1569
# With no old_root_keys we can just shortcut and be ready
1570
# for _flush_new_queue
1571
self._new_queue = list(self._new_root_keys)
1573
old_chks_to_enqueue = self._read_old_roots()
1574
# filter out any root keys that are already known to be uninteresting
1575
new_keys = set(self._new_root_keys).difference(self._all_old_chks)
1576
# These are prefixes that are present in new_keys that we are
1578
new_prefixes = set()
1579
# We are about to yield all of these, so we don't want them getting
1580
# added a second time
1581
processed_new_refs = self._processed_new_refs
1582
processed_new_refs.update(new_keys)
1583
for record, node, prefix_refs, items in \
1584
self._read_nodes_from_store(new_keys):
1585
# At this level, we now know all the uninteresting references
1586
# So we filter and queue up whatever is remaining
1587
prefix_refs = [p_r for p_r in prefix_refs
1588
if p_r[1] not in self._all_old_chks
1589
and p_r[1] not in processed_new_refs]
1590
refs = [p_r[1] for p_r in prefix_refs]
1591
new_prefixes.update([p_r[0] for p_r in prefix_refs])
1592
self._new_queue.extend(refs)
1593
# TODO: We can potentially get multiple items here, however the
1594
# current design allows for this, as callers will do the work
1595
# to make the results unique. We might profile whether we
1596
# gain anything by ensuring unique return values for items
1597
# TODO: This might be a good time to cast to StaticTuple, as
1598
# self._new_item_queue will hold the contents of multiple
1599
# records for an extended lifetime
1600
new_items = [item for item in items
1601
if item not in self._all_old_items]
1602
self._new_item_queue.extend(new_items)
1603
new_prefixes.update([self._search_key_func(item[0])
1604
for item in new_items])
1605
processed_new_refs.update(refs)
1607
# For new_prefixes we have the full length prefixes queued up.
1608
# However, we also need possible prefixes. (If we have a known ref to
1609
# 'ab', then we also need to include 'a'.) So expand the
1610
# new_prefixes to include all shorter prefixes
1611
for prefix in list(new_prefixes):
1612
new_prefixes.update([prefix[:i] for i in xrange(1, len(prefix))])
1613
self._enqueue_old(new_prefixes, old_chks_to_enqueue)
1615
def _flush_new_queue(self):
1616
# No need to maintain the heap invariant anymore, just pull things out
1618
refs = set(self._new_queue)
1619
self._new_queue = []
1620
# First pass, flush all interesting items and convert to using direct refs
1621
all_old_chks = self._all_old_chks
1622
processed_new_refs = self._processed_new_refs
1623
all_old_items = self._all_old_items
1624
new_items = [item for item in self._new_item_queue
1625
if item not in all_old_items]
1626
self._new_item_queue = []
1628
yield None, new_items
1629
refs = refs.difference(all_old_chks)
1630
processed_new_refs.update(refs)
1632
# TODO: Using a SimpleSet for self._processed_new_refs and
1633
# saved as much as 10MB of peak memory. However, it requires
1634
# implementing a non-pyrex version.
1636
next_refs_update = next_refs.update
1637
# Inlining _read_nodes_from_store improves 'bzr branch bzr.dev'
1638
# from 1m54s to 1m51s. Consider it.
1639
for record, _, p_refs, items in self._read_nodes_from_store(refs):
1641
# using the 'if' check saves about 145s => 141s, when
1642
# streaming initial branch of Launchpad data.
1643
items = [item for item in items
1644
if item not in all_old_items]
1646
next_refs_update([p_r[1] for p_r in p_refs])
1648
# set1.difference(set/dict) walks all of set1, and checks if it
1649
# exists in 'other'.
1650
# set1.difference(iterable) walks all of iterable, and does a
1651
# 'difference_update' on a clone of set1. Pick wisely based on the
1652
# expected sizes of objects.
1653
# in our case it is expected that 'new_refs' will always be quite
1655
next_refs = next_refs.difference(all_old_chks)
1656
next_refs = next_refs.difference(processed_new_refs)
1657
processed_new_refs.update(next_refs)
1660
def _process_next_old(self):
1661
# Since we don't filter uninteresting any further than during
1662
# _read_all_roots, process the whole queue in a single pass.
1663
refs = self._old_queue
1664
self._old_queue = []
1665
all_old_chks = self._all_old_chks
1666
for record, _, prefix_refs, items in self._read_nodes_from_store(refs):
1667
# TODO: Use StaticTuple here?
1668
self._all_old_items.update(items)
1669
refs = [r for _,r in prefix_refs if r not in all_old_chks]
1670
self._old_queue.extend(refs)
1671
all_old_chks.update(refs)
1673
def _process_queues(self):
1674
while self._old_queue:
1675
self._process_next_old()
1676
return self._flush_new_queue()
1679
for record in self._read_all_roots():
1681
for record, items in self._process_queues():
1685
def iter_interesting_nodes(store, interesting_root_keys,
1686
uninteresting_root_keys, pb=None):
1687
"""Given root keys, find interesting nodes.
1689
Evaluate nodes referenced by interesting_root_keys. Ones that are also
1690
referenced from uninteresting_root_keys are not considered interesting.
1692
:param interesting_root_keys: keys which should be part of the
1693
"interesting" nodes (which will be yielded)
1694
:param uninteresting_root_keys: keys which should be filtered out of the
1697
(interesting record, {interesting key:values})
1699
iterator = CHKMapDifference(store, interesting_root_keys,
1700
uninteresting_root_keys,
1701
search_key_func=store._search_key_func,
1703
return iterator.process()
1707
from bzrlib._chk_map_pyx import (
1710
_deserialise_leaf_node,
1711
_deserialise_internal_node,
1713
except ImportError, e:
1714
osutils.failed_to_load_extension(e)
1715
from bzrlib._chk_map_py import (
1718
_deserialise_leaf_node,
1719
_deserialise_internal_node,
1721
search_key_registry.register('hash-16-way', _search_key_16)
1722
search_key_registry.register('hash-255-way', _search_key_255)
1725
def _check_key(key):
1726
"""Helper function to assert that a key is properly formatted.
1728
This generally shouldn't be used in production code, but it can be helpful
1731
if type(key) is not StaticTuple:
1732
raise TypeError('key %r is not StaticTuple but %s' % (key, type(key)))
1734
raise ValueError('key %r should have length 1, not %d' % (key, len(key),))
1735
if type(key[0]) is not str:
1736
raise TypeError('key %r should hold a str, not %r'
1737
% (key, type(key[0])))
1738
if not key[0].startswith('sha1:'):
1739
raise ValueError('key %r should point to a sha1:' % (key,))
added
removed
bzrlib/chk_map.py
