1
# Copyright (C) 2008, 2009, 2010 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.
43
from bzrlib import lazy_import
44
lazy_import.lazy_import(globals(), """
57
from bzrlib.static_tuple import StaticTuple
60
# If each line is 50 bytes, and you have 255 internal pages, with 255-way fan
61
# out, it takes 3.1MB to cache the layer.
62
_PAGE_CACHE_SIZE = 4*1024*1024
63
# Per thread caches for 2 reasons:
64
# - in the server we may be serving very different content, so we get less
66
# - we avoid locking on every cache lookup.
67
_thread_caches = threading.local()
69
_thread_caches.page_cache = None
72
"""Get the per-thread page cache.
74
We need a function to do this because in a new thread the _thread_caches
75
threading.local object does not have the cache initialized yet.
77
page_cache = getattr(_thread_caches, 'page_cache', None)
78
if page_cache is None:
79
# We are caching bytes so len(value) is perfectly accurate
80
page_cache = lru_cache.LRUSizeCache(_PAGE_CACHE_SIZE)
81
_thread_caches.page_cache = page_cache
89
# If a ChildNode falls below this many bytes, we check for a remap
90
_INTERESTING_NEW_SIZE = 50
91
# If a ChildNode shrinks by more than this amount, we check for a remap
92
_INTERESTING_SHRINKAGE_LIMIT = 20
93
# If we delete more than this many nodes applying a delta, we check for a remap
94
_INTERESTING_DELETES_LIMIT = 5
97
def _search_key_plain(key):
98
"""Map the key tuple into a search string that just uses the key bytes."""
99
return '\x00'.join(key)
102
search_key_registry = registry.Registry()
103
search_key_registry.register('plain', _search_key_plain)
106
class CHKMap(object):
107
"""A persistent map from string to string backed by a CHK store."""
109
__slots__ = ('_store', '_root_node', '_search_key_func')
111
def __init__(self, store, root_key, search_key_func=None):
112
"""Create a CHKMap object.
114
:param store: The store the CHKMap is stored in.
115
:param root_key: The root key of the map. None to create an empty
117
:param search_key_func: A function mapping a key => bytes. These bytes
118
are then used by the internal nodes to split up leaf nodes into
122
if search_key_func is None:
123
search_key_func = _search_key_plain
124
self._search_key_func = search_key_func
126
self._root_node = LeafNode(search_key_func=search_key_func)
128
self._root_node = self._node_key(root_key)
130
def apply_delta(self, delta):
131
"""Apply a delta to the map.
133
:param delta: An iterable of old_key, new_key, new_value tuples.
134
If new_key is not None, then new_key->new_value is inserted
135
into the map; if old_key is not None, then the old mapping
136
of old_key is removed.
139
# Check preconditions first.
140
as_st = StaticTuple.from_sequence
141
new_items = set([as_st(key) for (old, key, value) in delta
142
if key is not None and old is None])
143
existing_new = list(self.iteritems(key_filter=new_items))
145
raise errors.InconsistentDeltaDelta(delta,
146
"New items are already in the map %r." % existing_new)
148
for old, new, value in delta:
149
if old is not None and old != new:
150
self.unmap(old, check_remap=False)
152
for old, new, value in delta:
155
if delete_count > _INTERESTING_DELETES_LIMIT:
156
trace.mutter("checking remap as %d deletions", delete_count)
160
def _ensure_root(self):
161
"""Ensure that the root node is an object not a key."""
162
if type(self._root_node) is StaticTuple:
163
# Demand-load the root
164
self._root_node = self._get_node(self._root_node)
166
def _get_node(self, node):
169
Note that this does not update the _items dict in objects containing a
170
reference to this node. As such it does not prevent subsequent IO being
173
:param node: A tuple key or node object.
174
:return: A node object.
176
if type(node) is StaticTuple:
177
bytes = self._read_bytes(node)
178
return _deserialise(bytes, node,
179
search_key_func=self._search_key_func)
183
def _read_bytes(self, key):
185
return _get_cache()[key]
187
stream = self._store.get_record_stream([key], 'unordered', True)
188
bytes = stream.next().get_bytes_as('fulltext')
189
_get_cache()[key] = bytes
192
def _dump_tree(self, include_keys=False):
193
"""Return the tree in a string representation."""
195
res = self._dump_tree_node(self._root_node, prefix='', indent='',
196
include_keys=include_keys)
197
res.append('') # Give a trailing '\n'
198
return '\n'.join(res)
200
def _dump_tree_node(self, node, prefix, indent, include_keys=True):
201
"""For this node and all children, generate a string representation."""
206
node_key = node.key()
207
if node_key is not None:
208
key_str = ' %s' % (node_key[0],)
211
result.append('%s%r %s%s' % (indent, prefix, node.__class__.__name__,
213
if type(node) is InternalNode:
214
# Trigger all child nodes to get loaded
215
list(node._iter_nodes(self._store))
216
for prefix, sub in sorted(node._items.iteritems()):
217
result.extend(self._dump_tree_node(sub, prefix, indent + ' ',
218
include_keys=include_keys))
220
for key, value in sorted(node._items.iteritems()):
221
# Don't use prefix nor indent here to line up when used in
222
# tests in conjunction with assertEqualDiff
223
result.append(' %r %r' % (tuple(key), value))
227
def from_dict(klass, store, initial_value, maximum_size=0, key_width=1,
228
search_key_func=None):
229
"""Create a CHKMap in store with initial_value as the content.
231
:param store: The store to record initial_value in, a VersionedFiles
232
object with 1-tuple keys supporting CHK key generation.
233
:param initial_value: A dict to store in store. Its keys and values
235
:param maximum_size: The maximum_size rule to apply to nodes. This
236
determines the size at which no new data is added to a single node.
237
:param key_width: The number of elements in each key_tuple being stored
239
:param search_key_func: A function mapping a key => bytes. These bytes
240
are then used by the internal nodes to split up leaf nodes into
242
:return: The root chk of the resulting CHKMap.
244
root_key = klass._create_directly(store, initial_value,
245
maximum_size=maximum_size, key_width=key_width,
246
search_key_func=search_key_func)
247
if type(root_key) is not StaticTuple:
248
raise AssertionError('we got a %s instead of a StaticTuple'
253
def _create_via_map(klass, store, initial_value, maximum_size=0,
254
key_width=1, search_key_func=None):
255
result = klass(store, None, search_key_func=search_key_func)
256
result._root_node.set_maximum_size(maximum_size)
257
result._root_node._key_width = key_width
259
for key, value in initial_value.items():
260
delta.append((None, key, value))
261
root_key = result.apply_delta(delta)
265
def _create_directly(klass, store, initial_value, maximum_size=0,
266
key_width=1, search_key_func=None):
267
node = LeafNode(search_key_func=search_key_func)
268
node.set_maximum_size(maximum_size)
269
node._key_width = key_width
270
as_st = StaticTuple.from_sequence
271
node._items = dict([(as_st(key), val) for key, val
272
in initial_value.iteritems()])
273
node._raw_size = sum([node._key_value_len(key, value)
274
for key,value in node._items.iteritems()])
275
node._len = len(node._items)
276
node._compute_search_prefix()
277
node._compute_serialised_prefix()
280
and node._current_size() > maximum_size):
281
prefix, node_details = node._split(store)
282
if len(node_details) == 1:
283
raise AssertionError('Failed to split using node._split')
284
node = InternalNode(prefix, search_key_func=search_key_func)
285
node.set_maximum_size(maximum_size)
286
node._key_width = key_width
287
for split, subnode in node_details:
288
node.add_node(split, subnode)
289
keys = list(node.serialise(store))
292
def iter_changes(self, basis):
293
"""Iterate over the changes between basis and self.
295
:return: An iterator of tuples: (key, old_value, new_value). Old_value
296
is None for keys only in self; new_value is None for keys only in
300
# Read both trees in lexographic, highest-first order.
301
# Any identical nodes we skip
302
# Any unique prefixes we output immediately.
303
# values in a leaf node are treated as single-value nodes in the tree
304
# which allows them to be not-special-cased. We know to output them
305
# because their value is a string, not a key(tuple) or node.
307
# corner cases to beware of when considering this function:
308
# *) common references are at different heights.
309
# consider two trees:
310
# {'a': LeafNode={'aaa':'foo', 'aab':'bar'}, 'b': LeafNode={'b'}}
311
# {'a': InternalNode={'aa':LeafNode={'aaa':'foo', 'aab':'bar'},
312
# 'ab':LeafNode={'ab':'bar'}}
313
# 'b': LeafNode={'b'}}
314
# the node with aaa/aab will only be encountered in the second tree
315
# after reading the 'a' subtree, but it is encountered in the first
316
# tree immediately. Variations on this may have read internal nodes
317
# like this. we want to cut the entire pending subtree when we
318
# realise we have a common node. For this we use a list of keys -
319
# the path to a node - and check the entire path is clean as we
321
if self._node_key(self._root_node) == self._node_key(basis._root_node):
325
excluded_keys = set()
326
self_node = self._root_node
327
basis_node = basis._root_node
328
# A heap, each element is prefix, node(tuple/NodeObject/string),
329
# key_path (a list of tuples, tail-sharing down the tree.)
332
def process_node(node, path, a_map, pending):
333
# take a node and expand it
334
node = a_map._get_node(node)
335
if type(node) == LeafNode:
336
path = (node._key, path)
337
for key, value in node._items.items():
338
# For a LeafNode, the key is a serialized_key, rather than
339
# a search_key, but the heap is using search_keys
340
search_key = node._search_key_func(key)
341
heapq.heappush(pending, (search_key, key, value, path))
343
# type(node) == InternalNode
344
path = (node._key, path)
345
for prefix, child in node._items.items():
346
heapq.heappush(pending, (prefix, None, child, path))
347
def process_common_internal_nodes(self_node, basis_node):
348
self_items = set(self_node._items.items())
349
basis_items = set(basis_node._items.items())
350
path = (self_node._key, None)
351
for prefix, child in self_items - basis_items:
352
heapq.heappush(self_pending, (prefix, None, child, path))
353
path = (basis_node._key, None)
354
for prefix, child in basis_items - self_items:
355
heapq.heappush(basis_pending, (prefix, None, child, path))
356
def process_common_leaf_nodes(self_node, basis_node):
357
self_items = set(self_node._items.items())
358
basis_items = set(basis_node._items.items())
359
path = (self_node._key, None)
360
for key, value in self_items - basis_items:
361
prefix = self._search_key_func(key)
362
heapq.heappush(self_pending, (prefix, key, value, path))
363
path = (basis_node._key, None)
364
for key, value in basis_items - self_items:
365
prefix = basis._search_key_func(key)
366
heapq.heappush(basis_pending, (prefix, key, value, path))
367
def process_common_prefix_nodes(self_node, self_path,
368
basis_node, basis_path):
369
# Would it be more efficient if we could request both at the same
371
self_node = self._get_node(self_node)
372
basis_node = basis._get_node(basis_node)
373
if (type(self_node) == InternalNode
374
and type(basis_node) == InternalNode):
375
# Matching internal nodes
376
process_common_internal_nodes(self_node, basis_node)
377
elif (type(self_node) == LeafNode
378
and type(basis_node) == LeafNode):
379
process_common_leaf_nodes(self_node, basis_node)
381
process_node(self_node, self_path, self, self_pending)
382
process_node(basis_node, basis_path, basis, basis_pending)
383
process_common_prefix_nodes(self_node, None, basis_node, None)
386
excluded_keys = set()
387
def check_excluded(key_path):
388
# Note that this is N^2, it depends on us trimming trees
389
# aggressively to not become slow.
390
# A better implementation would probably have a reverse map
391
# back to the children of a node, and jump straight to it when
392
# a common node is detected, the proceed to remove the already
393
# pending children. bzrlib.graph has a searcher module with a
395
while key_path is not None:
396
key, key_path = key_path
397
if key in excluded_keys:
402
while self_pending or basis_pending:
405
# self is exhausted: output remainder of basis
406
for prefix, key, node, path in basis_pending:
407
if check_excluded(path):
409
node = basis._get_node(node)
412
yield (key, node, None)
414
# subtree - fastpath the entire thing.
415
for key, value in node.iteritems(basis._store):
416
yield (key, value, None)
418
elif not basis_pending:
419
# basis is exhausted: output remainder of self.
420
for prefix, key, node, path in self_pending:
421
if check_excluded(path):
423
node = self._get_node(node)
426
yield (key, None, node)
428
# subtree - fastpath the entire thing.
429
for key, value in node.iteritems(self._store):
430
yield (key, None, value)
433
# XXX: future optimisation - yield the smaller items
434
# immediately rather than pushing everything on/off the
435
# heaps. Applies to both internal nodes and leafnodes.
436
if self_pending[0][0] < basis_pending[0][0]:
438
prefix, key, node, path = heapq.heappop(self_pending)
439
if check_excluded(path):
443
yield (key, None, node)
445
process_node(node, path, self, self_pending)
447
elif self_pending[0][0] > basis_pending[0][0]:
449
prefix, key, node, path = heapq.heappop(basis_pending)
450
if check_excluded(path):
454
yield (key, node, None)
456
process_node(node, path, basis, basis_pending)
459
# common prefix: possibly expand both
460
if self_pending[0][1] is None:
465
if basis_pending[0][1] is None:
470
if not read_self and not read_basis:
471
# compare a common value
472
self_details = heapq.heappop(self_pending)
473
basis_details = heapq.heappop(basis_pending)
474
if self_details[2] != basis_details[2]:
475
yield (self_details[1],
476
basis_details[2], self_details[2])
478
# At least one side wasn't a simple value
479
if (self._node_key(self_pending[0][2]) ==
480
self._node_key(basis_pending[0][2])):
481
# Identical pointers, skip (and don't bother adding to
482
# excluded, it won't turn up again.
483
heapq.heappop(self_pending)
484
heapq.heappop(basis_pending)
486
# Now we need to expand this node before we can continue
487
if read_self and read_basis:
488
# Both sides start with the same prefix, so process
490
self_prefix, _, self_node, self_path = heapq.heappop(
492
basis_prefix, _, basis_node, basis_path = heapq.heappop(
494
if self_prefix != basis_prefix:
495
raise AssertionError(
496
'%r != %r' % (self_prefix, basis_prefix))
497
process_common_prefix_nodes(
498
self_node, self_path,
499
basis_node, basis_path)
502
prefix, key, node, path = heapq.heappop(self_pending)
503
if check_excluded(path):
505
process_node(node, path, self, self_pending)
507
prefix, key, node, path = heapq.heappop(basis_pending)
508
if check_excluded(path):
510
process_node(node, path, basis, basis_pending)
513
def iteritems(self, key_filter=None):
514
"""Iterate over the entire CHKMap's contents."""
516
if key_filter is not None:
517
as_st = StaticTuple.from_sequence
518
key_filter = [as_st(key) for key in key_filter]
519
return self._root_node.iteritems(self._store, key_filter=key_filter)
522
"""Return the key for this map."""
523
if type(self._root_node) is StaticTuple:
524
return self._root_node
526
return self._root_node._key
530
return len(self._root_node)
532
def map(self, key, value):
533
"""Map a key tuple to value.
535
:param key: A key to map.
536
:param value: The value to assign to key.
538
key = StaticTuple.from_sequence(key)
539
# Need a root object.
541
prefix, node_details = self._root_node.map(self._store, key, value)
542
if len(node_details) == 1:
543
self._root_node = node_details[0][1]
545
self._root_node = InternalNode(prefix,
546
search_key_func=self._search_key_func)
547
self._root_node.set_maximum_size(node_details[0][1].maximum_size)
548
self._root_node._key_width = node_details[0][1]._key_width
549
for split, node in node_details:
550
self._root_node.add_node(split, node)
552
def _node_key(self, node):
553
"""Get the key for a node whether it's a tuple or node."""
554
if type(node) is tuple:
555
node = StaticTuple.from_sequence(node)
556
if type(node) is StaticTuple:
561
def unmap(self, key, check_remap=True):
562
"""remove key from the map."""
563
key = StaticTuple.from_sequence(key)
565
if type(self._root_node) is InternalNode:
566
unmapped = self._root_node.unmap(self._store, key,
567
check_remap=check_remap)
569
unmapped = self._root_node.unmap(self._store, key)
570
self._root_node = unmapped
572
def _check_remap(self):
573
"""Check if nodes can be collapsed."""
575
if type(self._root_node) is InternalNode:
576
self._root_node._check_remap(self._store)
579
"""Save the map completely.
581
:return: The key of the root node.
583
if type(self._root_node) is StaticTuple:
585
return self._root_node
586
keys = list(self._root_node.serialise(self._store))
591
"""Base class defining the protocol for CHK Map nodes.
593
:ivar _raw_size: The total size of the serialized key:value data, before
594
adding the header bytes, and without prefix compression.
597
__slots__ = ('_key', '_len', '_maximum_size', '_key_width',
598
'_raw_size', '_items', '_search_prefix', '_search_key_func'
601
def __init__(self, key_width=1):
604
:param key_width: The width of keys for this node.
607
# Current number of elements
609
self._maximum_size = 0
610
self._key_width = key_width
611
# current size in bytes
613
# The pointers/values this node has - meaning defined by child classes.
615
# The common search prefix
616
self._search_prefix = None
619
items_str = str(sorted(self._items))
620
if len(items_str) > 20:
621
items_str = items_str[:16] + '...]'
622
return '%s(key:%s len:%s size:%s max:%s prefix:%s items:%s)' % (
623
self.__class__.__name__, self._key, self._len, self._raw_size,
624
self._maximum_size, self._search_prefix, items_str)
633
def maximum_size(self):
634
"""What is the upper limit for adding references to a node."""
635
return self._maximum_size
637
def set_maximum_size(self, new_size):
638
"""Set the size threshold for nodes.
640
:param new_size: The size at which no data is added to a node. 0 for
643
self._maximum_size = new_size
646
def common_prefix(cls, prefix, key):
647
"""Given 2 strings, return the longest prefix common to both.
649
:param prefix: This has been the common prefix for other keys, so it is
650
more likely to be the common prefix in this case as well.
651
:param key: Another string to compare to
653
if key.startswith(prefix):
656
# Is there a better way to do this?
657
for pos, (left, right) in enumerate(zip(prefix, key)):
661
common = prefix[:pos+1]
665
def common_prefix_for_keys(cls, keys):
666
"""Given a list of keys, find their common prefix.
668
:param keys: An iterable of strings.
669
:return: The longest common prefix of all keys.
673
if common_prefix is None:
676
common_prefix = cls.common_prefix(common_prefix, key)
677
if not common_prefix:
678
# if common_prefix is the empty string, then we know it won't
684
# Singleton indicating we have not computed _search_prefix yet
687
class LeafNode(Node):
688
"""A node containing actual key:value pairs.
690
:ivar _items: A dict of key->value items. The key is in tuple form.
691
:ivar _size: The number of bytes that would be used by serializing all of
695
__slots__ = ('_common_serialised_prefix',)
697
def __init__(self, search_key_func=None):
699
# All of the keys in this leaf node share this common prefix
700
self._common_serialised_prefix = None
701
if search_key_func is None:
702
self._search_key_func = _search_key_plain
704
self._search_key_func = search_key_func
707
items_str = str(sorted(self._items))
708
if len(items_str) > 20:
709
items_str = items_str[:16] + '...]'
711
'%s(key:%s len:%s size:%s max:%s prefix:%s keywidth:%s items:%s)' \
712
% (self.__class__.__name__, self._key, self._len, self._raw_size,
713
self._maximum_size, self._search_prefix, self._key_width, items_str)
715
def _current_size(self):
716
"""Answer the current serialised size of this node.
718
This differs from self._raw_size in that it includes the bytes used for
721
if self._common_serialised_prefix is None:
725
# We will store a single string with the common prefix
726
# And then that common prefix will not be stored in any of the
728
prefix_len = len(self._common_serialised_prefix)
729
bytes_for_items = (self._raw_size - (prefix_len * self._len))
730
return (9 # 'chkleaf:\n'
731
+ len(str(self._maximum_size)) + 1
732
+ len(str(self._key_width)) + 1
733
+ len(str(self._len)) + 1
738
def deserialise(klass, bytes, key, search_key_func=None):
739
"""Deserialise bytes, with key key, into a LeafNode.
741
:param bytes: The bytes of the node.
742
:param key: The key that the serialised node has.
744
key = static_tuple.expect_static_tuple(key)
745
return _deserialise_leaf_node(bytes, key,
746
search_key_func=search_key_func)
748
def iteritems(self, store, key_filter=None):
749
"""Iterate over items in the node.
751
:param key_filter: A filter to apply to the node. It should be a
752
list/set/dict or similar repeatedly iterable container.
754
if key_filter is not None:
755
# Adjust the filter - short elements go to a prefix filter. All
756
# other items are looked up directly.
757
# XXX: perhaps defaultdict? Profiling<rinse and repeat>
759
for key in key_filter:
760
if len(key) == self._key_width:
761
# This filter is meant to match exactly one key, yield it
764
yield key, self._items[key]
766
# This key is not present in this map, continue
769
# Short items, we need to match based on a prefix
770
length_filter = filters.setdefault(len(key), set())
771
length_filter.add(key)
773
filters = filters.items()
774
for item in self._items.iteritems():
775
for length, length_filter in filters:
776
if item[0][:length] in length_filter:
780
for item in self._items.iteritems():
783
def _key_value_len(self, key, value):
784
# TODO: Should probably be done without actually joining the key, but
785
# then that can be done via the C extension
786
return (len(self._serialise_key(key)) + 1
787
+ len(str(value.count('\n'))) + 1
790
def _search_key(self, key):
791
return self._search_key_func(key)
793
def _map_no_split(self, key, value):
794
"""Map a key to a value.
796
This assumes either the key does not already exist, or you have already
797
removed its size and length from self.
799
:return: True if adding this node should cause us to split.
801
self._items[key] = value
802
self._raw_size += self._key_value_len(key, value)
804
serialised_key = self._serialise_key(key)
805
if self._common_serialised_prefix is None:
806
self._common_serialised_prefix = serialised_key
808
self._common_serialised_prefix = self.common_prefix(
809
self._common_serialised_prefix, serialised_key)
810
search_key = self._search_key(key)
811
if self._search_prefix is _unknown:
812
self._compute_search_prefix()
813
if self._search_prefix is None:
814
self._search_prefix = search_key
816
self._search_prefix = self.common_prefix(
817
self._search_prefix, search_key)
819
and self._maximum_size
820
and self._current_size() > self._maximum_size):
821
# Check to see if all of the search_keys for this node are
822
# identical. We allow the node to grow under that circumstance
823
# (we could track this as common state, but it is infrequent)
824
if (search_key != self._search_prefix
825
or not self._are_search_keys_identical()):
829
def _split(self, store):
830
"""We have overflowed.
832
Split this node into multiple LeafNodes, return it up the stack so that
833
the next layer creates a new InternalNode and references the new nodes.
835
:return: (common_serialised_prefix, [(node_serialised_prefix, node)])
837
if self._search_prefix is _unknown:
838
raise AssertionError('Search prefix must be known')
839
common_prefix = self._search_prefix
840
split_at = len(common_prefix) + 1
842
for key, value in self._items.iteritems():
843
search_key = self._search_key(key)
844
prefix = search_key[:split_at]
845
# TODO: Generally only 1 key can be exactly the right length,
846
# which means we can only have 1 key in the node pointed
847
# at by the 'prefix\0' key. We might want to consider
848
# folding it into the containing InternalNode rather than
849
# having a fixed length-1 node.
850
# Note this is probably not true for hash keys, as they
851
# may get a '\00' node anywhere, but won't have keys of
853
if len(prefix) < split_at:
854
prefix += '\x00'*(split_at - len(prefix))
855
if prefix not in result:
856
node = LeafNode(search_key_func=self._search_key_func)
857
node.set_maximum_size(self._maximum_size)
858
node._key_width = self._key_width
859
result[prefix] = node
861
node = result[prefix]
862
sub_prefix, node_details = node.map(store, key, value)
863
if len(node_details) > 1:
864
if prefix != sub_prefix:
865
# This node has been split and is now found via a different
868
new_node = InternalNode(sub_prefix,
869
search_key_func=self._search_key_func)
870
new_node.set_maximum_size(self._maximum_size)
871
new_node._key_width = self._key_width
872
for split, node in node_details:
873
new_node.add_node(split, node)
874
result[prefix] = new_node
875
return common_prefix, result.items()
877
def map(self, store, key, value):
878
"""Map key to value."""
879
if key in self._items:
880
self._raw_size -= self._key_value_len(key, self._items[key])
883
if self._map_no_split(key, value):
884
return self._split(store)
886
if self._search_prefix is _unknown:
887
raise AssertionError('%r must be known' % self._search_prefix)
888
return self._search_prefix, [("", self)]
890
_serialise_key = '\x00'.join
892
def serialise(self, store):
893
"""Serialise the LeafNode to store.
895
:param store: A VersionedFiles honouring the CHK extensions.
896
:return: An iterable of the keys inserted by this operation.
898
lines = ["chkleaf:\n"]
899
lines.append("%d\n" % self._maximum_size)
900
lines.append("%d\n" % self._key_width)
901
lines.append("%d\n" % self._len)
902
if self._common_serialised_prefix is None:
904
if len(self._items) != 0:
905
raise AssertionError('If _common_serialised_prefix is None'
906
' we should have no items')
908
lines.append('%s\n' % (self._common_serialised_prefix,))
909
prefix_len = len(self._common_serialised_prefix)
910
for key, value in sorted(self._items.items()):
911
# Always add a final newline
912
value_lines = osutils.chunks_to_lines([value + '\n'])
913
serialized = "%s\x00%s\n" % (self._serialise_key(key),
915
if not serialized.startswith(self._common_serialised_prefix):
916
raise AssertionError('We thought the common prefix was %r'
917
' but entry %r does not have it in common'
918
% (self._common_serialised_prefix, serialized))
919
lines.append(serialized[prefix_len:])
920
lines.extend(value_lines)
921
sha1, _, _ = store.add_lines((None,), (), lines)
922
self._key = StaticTuple("sha1:" + sha1,).intern()
923
bytes = ''.join(lines)
924
if len(bytes) != self._current_size():
925
raise AssertionError('Invalid _current_size')
926
_get_cache().add(self._key, bytes)
930
"""Return the references to other CHK's held by this node."""
933
def _compute_search_prefix(self):
934
"""Determine the common search prefix for all keys in this node.
936
:return: A bytestring of the longest search key prefix that is
937
unique within this node.
939
search_keys = [self._search_key_func(key) for key in self._items]
940
self._search_prefix = self.common_prefix_for_keys(search_keys)
941
return self._search_prefix
943
def _are_search_keys_identical(self):
944
"""Check to see if the search keys for all entries are the same.
946
When using a hash as the search_key it is possible for non-identical
947
keys to collide. If that happens enough, we may try overflow a
948
LeafNode, but as all are collisions, we must not split.
950
common_search_key = None
951
for key in self._items:
952
search_key = self._search_key(key)
953
if common_search_key is None:
954
common_search_key = search_key
955
elif search_key != common_search_key:
959
def _compute_serialised_prefix(self):
960
"""Determine the common prefix for serialised keys in this node.
962
:return: A bytestring of the longest serialised key prefix that is
963
unique within this node.
965
serialised_keys = [self._serialise_key(key) for key in self._items]
966
self._common_serialised_prefix = self.common_prefix_for_keys(
968
return self._common_serialised_prefix
970
def unmap(self, store, key):
971
"""Unmap key from the node."""
973
self._raw_size -= self._key_value_len(key, self._items[key])
975
trace.mutter("key %s not found in %r", key, self._items)
980
# Recompute from scratch
981
self._compute_search_prefix()
982
self._compute_serialised_prefix()
986
class InternalNode(Node):
987
"""A node that contains references to other nodes.
989
An InternalNode is responsible for mapping search key prefixes to child
992
:ivar _items: serialised_key => node dictionary. node may be a tuple,
993
LeafNode or InternalNode.
996
__slots__ = ('_node_width',)
998
def __init__(self, prefix='', search_key_func=None):
1000
# The size of an internalnode with default values and no children.
1001
# How many octets key prefixes within this node are.
1002
self._node_width = 0
1003
self._search_prefix = prefix
1004
if search_key_func is None:
1005
self._search_key_func = _search_key_plain
1007
self._search_key_func = search_key_func
1009
def add_node(self, prefix, node):
1010
"""Add a child node with prefix prefix, and node node.
1012
:param prefix: The search key prefix for node.
1013
:param node: The node being added.
1015
if self._search_prefix is None:
1016
raise AssertionError("_search_prefix should not be None")
1017
if not prefix.startswith(self._search_prefix):
1018
raise AssertionError("prefixes mismatch: %s must start with %s"
1019
% (prefix,self._search_prefix))
1020
if len(prefix) != len(self._search_prefix) + 1:
1021
raise AssertionError("prefix wrong length: len(%s) is not %d" %
1022
(prefix, len(self._search_prefix) + 1))
1023
self._len += len(node)
1024
if not len(self._items):
1025
self._node_width = len(prefix)
1026
if self._node_width != len(self._search_prefix) + 1:
1027
raise AssertionError("node width mismatch: %d is not %d" %
1028
(self._node_width, len(self._search_prefix) + 1))
1029
self._items[prefix] = node
1032
def _current_size(self):
1033
"""Answer the current serialised size of this node."""
1034
return (self._raw_size + len(str(self._len)) + len(str(self._key_width)) +
1035
len(str(self._maximum_size)))
1038
def deserialise(klass, bytes, key, search_key_func=None):
1039
"""Deserialise bytes to an InternalNode, with key key.
1041
:param bytes: The bytes of the node.
1042
:param key: The key that the serialised node has.
1043
:return: An InternalNode instance.
1045
key = static_tuple.expect_static_tuple(key)
1046
return _deserialise_internal_node(bytes, key,
1047
search_key_func=search_key_func)
1049
def iteritems(self, store, key_filter=None):
1050
for node, node_filter in self._iter_nodes(store, key_filter=key_filter):
1051
for item in node.iteritems(store, key_filter=node_filter):
1054
def _iter_nodes(self, store, key_filter=None, batch_size=None):
1055
"""Iterate over node objects which match key_filter.
1057
:param store: A store to use for accessing content.
1058
:param key_filter: A key filter to filter nodes. Only nodes that might
1059
contain a key in key_filter will be returned.
1060
:param batch_size: If not None, then we will return the nodes that had
1061
to be read using get_record_stream in batches, rather than reading
1063
:return: An iterable of nodes. This function does not have to be fully
1064
consumed. (There will be no pending I/O when items are being returned.)
1066
# Map from chk key ('sha1:...',) to (prefix, key_filter)
1067
# prefix is the key in self._items to use, key_filter is the key_filter
1068
# entries that would match this node
1071
if key_filter is None:
1072
# yielding all nodes, yield whatever we have, and queue up a read
1073
# for whatever we are missing
1075
for prefix, node in self._items.iteritems():
1076
if node.__class__ is StaticTuple:
1077
keys[node] = (prefix, None)
1080
elif len(key_filter) == 1:
1081
# Technically, this path could also be handled by the first check
1082
# in 'self._node_width' in length_filters. However, we can handle
1083
# this case without spending any time building up the
1084
# prefix_to_keys, etc state.
1086
# This is a bit ugly, but TIMEIT showed it to be by far the fastest
1087
# 0.626us list(key_filter)[0]
1088
# is a func() for list(), 2 mallocs, and a getitem
1089
# 0.489us [k for k in key_filter][0]
1090
# still has the mallocs, avoids the func() call
1091
# 0.350us iter(key_filter).next()
1092
# has a func() call, and mallocs an iterator
1093
# 0.125us for key in key_filter: pass
1094
# no func() overhead, might malloc an iterator
1095
# 0.105us for key in key_filter: break
1096
# no func() overhead, might malloc an iterator, probably
1097
# avoids checking an 'else' clause as part of the for
1098
for key in key_filter:
1100
search_prefix = self._search_prefix_filter(key)
1101
if len(search_prefix) == self._node_width:
1102
# This item will match exactly, so just do a dict lookup, and
1103
# see what we can return
1106
node = self._items[search_prefix]
1108
# A given key can only match 1 child node, if it isn't
1109
# there, then we can just return nothing
1111
if node.__class__ is StaticTuple:
1112
keys[node] = (search_prefix, [key])
1114
# This is loaded, and the only thing that can match,
1119
# First, convert all keys into a list of search prefixes
1120
# Aggregate common prefixes, and track the keys they come from
1123
for key in key_filter:
1124
search_prefix = self._search_prefix_filter(key)
1125
length_filter = length_filters.setdefault(
1126
len(search_prefix), set())
1127
length_filter.add(search_prefix)
1128
prefix_to_keys.setdefault(search_prefix, []).append(key)
1130
if (self._node_width in length_filters
1131
and len(length_filters) == 1):
1132
# all of the search prefixes match exactly _node_width. This
1133
# means that everything is an exact match, and we can do a
1134
# lookup into self._items, rather than iterating over the items
1136
search_prefixes = length_filters[self._node_width]
1137
for search_prefix in search_prefixes:
1139
node = self._items[search_prefix]
1141
# We can ignore this one
1143
node_key_filter = prefix_to_keys[search_prefix]
1144
if node.__class__ is StaticTuple:
1145
keys[node] = (search_prefix, node_key_filter)
1147
yield node, node_key_filter
1149
# The slow way. We walk every item in self._items, and check to
1150
# see if there are any matches
1151
length_filters = length_filters.items()
1152
for prefix, node in self._items.iteritems():
1153
node_key_filter = []
1154
for length, length_filter in length_filters:
1155
sub_prefix = prefix[:length]
1156
if sub_prefix in length_filter:
1157
node_key_filter.extend(prefix_to_keys[sub_prefix])
1158
if node_key_filter: # this key matched something, yield it
1159
if node.__class__ is StaticTuple:
1160
keys[node] = (prefix, node_key_filter)
1162
yield node, node_key_filter
1164
# Look in the page cache for some more bytes
1168
bytes = _get_cache()[key]
1172
node = _deserialise(bytes, key,
1173
search_key_func=self._search_key_func)
1174
prefix, node_key_filter = keys[key]
1175
self._items[prefix] = node
1177
yield node, node_key_filter
1178
for key in found_keys:
1181
# demand load some pages.
1182
if batch_size is None:
1183
# Read all the keys in
1184
batch_size = len(keys)
1185
key_order = list(keys)
1186
for batch_start in range(0, len(key_order), batch_size):
1187
batch = key_order[batch_start:batch_start + batch_size]
1188
# We have to fully consume the stream so there is no pending
1189
# I/O, so we buffer the nodes for now.
1190
stream = store.get_record_stream(batch, 'unordered', True)
1191
node_and_filters = []
1192
for record in stream:
1193
bytes = record.get_bytes_as('fulltext')
1194
node = _deserialise(bytes, record.key,
1195
search_key_func=self._search_key_func)
1196
prefix, node_key_filter = keys[record.key]
1197
node_and_filters.append((node, node_key_filter))
1198
self._items[prefix] = node
1199
_get_cache().add(record.key, bytes)
1200
for info in node_and_filters:
1203
def map(self, store, key, value):
1204
"""Map key to value."""
1205
if not len(self._items):
1206
raise AssertionError("can't map in an empty InternalNode.")
1207
search_key = self._search_key(key)
1208
if self._node_width != len(self._search_prefix) + 1:
1209
raise AssertionError("node width mismatch: %d is not %d" %
1210
(self._node_width, len(self._search_prefix) + 1))
1211
if not search_key.startswith(self._search_prefix):
1212
# This key doesn't fit in this index, so we need to split at the
1213
# point where it would fit, insert self into that internal node,
1214
# and then map this key into that node.
1215
new_prefix = self.common_prefix(self._search_prefix,
1217
new_parent = InternalNode(new_prefix,
1218
search_key_func=self._search_key_func)
1219
new_parent.set_maximum_size(self._maximum_size)
1220
new_parent._key_width = self._key_width
1221
new_parent.add_node(self._search_prefix[:len(new_prefix)+1],
1223
return new_parent.map(store, key, value)
1224
children = [node for node, _
1225
in self._iter_nodes(store, key_filter=[key])]
1230
child = self._new_child(search_key, LeafNode)
1231
old_len = len(child)
1232
if type(child) is LeafNode:
1233
old_size = child._current_size()
1236
prefix, node_details = child.map(store, key, value)
1237
if len(node_details) == 1:
1238
# child may have shrunk, or might be a new node
1239
child = node_details[0][1]
1240
self._len = self._len - old_len + len(child)
1241
self._items[search_key] = child
1244
if type(child) is LeafNode:
1245
if old_size is None:
1246
# The old node was an InternalNode which means it has now
1247
# collapsed, so we need to check if it will chain to a
1248
# collapse at this level.
1249
trace.mutter("checking remap as InternalNode -> LeafNode")
1250
new_node = self._check_remap(store)
1252
# If the LeafNode has shrunk in size, we may want to run
1253
# a remap check. Checking for a remap is expensive though
1254
# and the frequency of a successful remap is very low.
1255
# Shrinkage by small amounts is common, so we only do the
1256
# remap check if the new_size is low or the shrinkage
1257
# amount is over a configurable limit.
1258
new_size = child._current_size()
1259
shrinkage = old_size - new_size
1260
if (shrinkage > 0 and new_size < _INTERESTING_NEW_SIZE
1261
or shrinkage > _INTERESTING_SHRINKAGE_LIMIT):
1263
"checking remap as size shrunk by %d to be %d",
1264
shrinkage, new_size)
1265
new_node = self._check_remap(store)
1266
if new_node._search_prefix is None:
1267
raise AssertionError("_search_prefix should not be None")
1268
return new_node._search_prefix, [('', new_node)]
1269
# child has overflown - create a new intermediate node.
1270
# XXX: This is where we might want to try and expand our depth
1271
# to refer to more bytes of every child (which would give us
1272
# multiple pointers to child nodes, but less intermediate nodes)
1273
child = self._new_child(search_key, InternalNode)
1274
child._search_prefix = prefix
1275
for split, node in node_details:
1276
child.add_node(split, node)
1277
self._len = self._len - old_len + len(child)
1279
return self._search_prefix, [("", self)]
1281
def _new_child(self, search_key, klass):
1282
"""Create a new child node of type klass."""
1284
child.set_maximum_size(self._maximum_size)
1285
child._key_width = self._key_width
1286
child._search_key_func = self._search_key_func
1287
self._items[search_key] = child
1290
def serialise(self, store):
1291
"""Serialise the node to store.
1293
:param store: A VersionedFiles honouring the CHK extensions.
1294
:return: An iterable of the keys inserted by this operation.
1296
for node in self._items.itervalues():
1297
if type(node) is StaticTuple:
1298
# Never deserialised.
1300
if node._key is not None:
1303
for key in node.serialise(store):
1305
lines = ["chknode:\n"]
1306
lines.append("%d\n" % self._maximum_size)
1307
lines.append("%d\n" % self._key_width)
1308
lines.append("%d\n" % self._len)
1309
if self._search_prefix is None:
1310
raise AssertionError("_search_prefix should not be None")
1311
lines.append('%s\n' % (self._search_prefix,))
1312
prefix_len = len(self._search_prefix)
1313
for prefix, node in sorted(self._items.items()):
1314
if type(node) is StaticTuple:
1318
serialised = "%s\x00%s\n" % (prefix, key)
1319
if not serialised.startswith(self._search_prefix):
1320
raise AssertionError("prefixes mismatch: %s must start with %s"
1321
% (serialised, self._search_prefix))
1322
lines.append(serialised[prefix_len:])
1323
sha1, _, _ = store.add_lines((None,), (), lines)
1324
self._key = StaticTuple("sha1:" + sha1,).intern()
1325
_get_cache().add(self._key, ''.join(lines))
1328
def _search_key(self, key):
1329
"""Return the serialised key for key in this node."""
1330
# search keys are fixed width. All will be self._node_width wide, so we
1332
return (self._search_key_func(key) + '\x00'*self._node_width)[:self._node_width]
1334
def _search_prefix_filter(self, key):
1335
"""Serialise key for use as a prefix filter in iteritems."""
1336
return self._search_key_func(key)[:self._node_width]
1338
def _split(self, offset):
1339
"""Split this node into smaller nodes starting at offset.
1341
:param offset: The offset to start the new child nodes at.
1342
:return: An iterable of (prefix, node) tuples. prefix is a byte
1343
prefix for reaching node.
1345
if offset >= self._node_width:
1346
for node in self._items.values():
1347
for result in node._split(offset):
1350
for key, node in self._items.items():
1354
"""Return the references to other CHK's held by this node."""
1355
if self._key is None:
1356
raise AssertionError("unserialised nodes have no refs.")
1358
for value in self._items.itervalues():
1359
if type(value) is StaticTuple:
1362
refs.append(value.key())
1365
def _compute_search_prefix(self, extra_key=None):
1366
"""Return the unique key prefix for this node.
1368
:return: A bytestring of the longest search key prefix that is
1369
unique within this node.
1371
self._search_prefix = self.common_prefix_for_keys(self._items)
1372
return self._search_prefix
1374
def unmap(self, store, key, check_remap=True):
1375
"""Remove key from this node and it's children."""
1376
if not len(self._items):
1377
raise AssertionError("can't unmap in an empty InternalNode.")
1378
children = [node for node, _
1379
in self._iter_nodes(store, key_filter=[key])]
1385
unmapped = child.unmap(store, key)
1387
search_key = self._search_key(key)
1388
if len(unmapped) == 0:
1389
# All child nodes are gone, remove the child:
1390
del self._items[search_key]
1393
# Stash the returned node
1394
self._items[search_key] = unmapped
1395
if len(self._items) == 1:
1396
# this node is no longer needed:
1397
return self._items.values()[0]
1398
if type(unmapped) is InternalNode:
1401
return self._check_remap(store)
1405
def _check_remap(self, store):
1406
"""Check if all keys contained by children fit in a single LeafNode.
1408
:param store: A store to use for reading more nodes
1409
:return: Either self, or a new LeafNode which should replace self.
1411
# Logic for how we determine when we need to rebuild
1412
# 1) Implicitly unmap() is removing a key which means that the child
1413
# nodes are going to be shrinking by some extent.
1414
# 2) If all children are LeafNodes, it is possible that they could be
1415
# combined into a single LeafNode, which can then completely replace
1416
# this internal node with a single LeafNode
1417
# 3) If *one* child is an InternalNode, we assume it has already done
1418
# all the work to determine that its children cannot collapse, and
1419
# we can then assume that those nodes *plus* the current nodes don't
1420
# have a chance of collapsing either.
1421
# So a very cheap check is to just say if 'unmapped' is an
1422
# InternalNode, we don't have to check further.
1424
# TODO: Another alternative is to check the total size of all known
1425
# LeafNodes. If there is some formula we can use to determine the
1426
# final size without actually having to read in any more
1427
# children, it would be nice to have. However, we have to be
1428
# careful with stuff like nodes that pull out the common prefix
1429
# of each key, as adding a new key can change the common prefix
1430
# and cause size changes greater than the length of one key.
1431
# So for now, we just add everything to a new Leaf until it
1432
# splits, as we know that will give the right answer
1433
new_leaf = LeafNode(search_key_func=self._search_key_func)
1434
new_leaf.set_maximum_size(self._maximum_size)
1435
new_leaf._key_width = self._key_width
1436
# A batch_size of 16 was chosen because:
1437
# a) In testing, a 4k page held 14 times. So if we have more than 16
1438
# leaf nodes we are unlikely to hold them in a single new leaf
1439
# node. This still allows for 1 round trip
1440
# b) With 16-way fan out, we can still do a single round trip
1441
# c) With 255-way fan out, we don't want to read all 255 and destroy
1442
# the page cache, just to determine that we really don't need it.
1443
for node, _ in self._iter_nodes(store, batch_size=16):
1444
if type(node) is InternalNode:
1445
# Without looking at any leaf nodes, we are sure
1447
for key, value in node._items.iteritems():
1448
if new_leaf._map_no_split(key, value):
1450
trace.mutter("remap generated a new LeafNode")
1454
def _deserialise(bytes, key, search_key_func):
1455
"""Helper for repositorydetails - convert bytes to a node."""
1456
if bytes.startswith("chkleaf:\n"):
1457
node = LeafNode.deserialise(bytes, key, search_key_func=search_key_func)
1458
elif bytes.startswith("chknode:\n"):
1459
node = InternalNode.deserialise(bytes, key,
1460
search_key_func=search_key_func)
1462
raise AssertionError("Unknown node type.")
1466
class CHKMapDifference(object):
1467
"""Iterate the stored pages and key,value pairs for (new - old).
1469
This class provides a generator over the stored CHK pages and the
1470
(key, value) pairs that are in any of the new maps and not in any of the
1473
Note that it may yield chk pages that are common (especially root nodes),
1474
but it won't yield (key,value) pairs that are common.
1477
def __init__(self, store, new_root_keys, old_root_keys,
1478
search_key_func, pb=None):
1479
# TODO: Should we add a StaticTuple barrier here? It would be nice to
1480
# force callers to use StaticTuple, because there will often be
1481
# lots of keys passed in here. And even if we cast it locally,
1482
# that just meanst that we will have *both* a StaticTuple and a
1483
# tuple() in memory, referring to the same object. (so a net
1484
# increase in memory, not a decrease.)
1486
self._new_root_keys = new_root_keys
1487
self._old_root_keys = old_root_keys
1489
# All uninteresting chks that we have seen. By the time they are added
1490
# here, they should be either fully ignored, or queued up for
1492
# TODO: This might grow to a large size if there are lots of merge
1493
# parents, etc. However, it probably doesn't scale to O(history)
1494
# like _processed_new_refs does.
1495
self._all_old_chks = set(self._old_root_keys)
1496
# All items that we have seen from the old_root_keys
1497
self._all_old_items = set()
1498
# These are interesting items which were either read, or already in the
1499
# interesting queue (so we don't need to walk them again)
1500
# TODO: processed_new_refs becomes O(all_chks), consider switching to
1502
self._processed_new_refs = set()
1503
self._search_key_func = search_key_func
1505
# The uninteresting and interesting nodes to be searched
1506
self._old_queue = []
1507
self._new_queue = []
1508
# Holds the (key, value) items found when processing the root nodes,
1509
# waiting for the uninteresting nodes to be walked
1510
self._new_item_queue = []
1513
def _read_nodes_from_store(self, keys):
1514
# We chose not to use _get_cache(), because we think in
1515
# terms of records to be yielded. Also, we expect to touch each page
1516
# only 1 time during this code. (We may want to evaluate saving the
1517
# raw bytes into the page cache, which would allow a working tree
1518
# update after the fetch to not have to read the bytes again.)
1519
as_st = StaticTuple.from_sequence
1520
stream = self._store.get_record_stream(keys, 'unordered', True)
1521
for record in stream:
1522
if self._pb is not None:
1524
if record.storage_kind == 'absent':
1525
raise errors.NoSuchRevision(self._store, record.key)
1526
bytes = record.get_bytes_as('fulltext')
1527
node = _deserialise(bytes, record.key,
1528
search_key_func=self._search_key_func)
1529
if type(node) is InternalNode:
1530
# Note we don't have to do node.refs() because we know that
1531
# there are no children that have been pushed into this node
1532
# Note: Using as_st() here seemed to save 1.2MB, which would
1533
# indicate that we keep 100k prefix_refs around while
1534
# processing. They *should* be shorter lived than that...
1535
# It does cost us ~10s of processing time
1536
#prefix_refs = [as_st(item) for item in node._items.iteritems()]
1537
prefix_refs = node._items.items()
1541
# Note: We don't use a StaticTuple here. Profiling showed a
1542
# minor memory improvement (0.8MB out of 335MB peak 0.2%)
1543
# But a significant slowdown (15s / 145s, or 10%)
1544
items = node._items.items()
1545
yield record, node, prefix_refs, items
1547
def _read_old_roots(self):
1548
old_chks_to_enqueue = []
1549
all_old_chks = self._all_old_chks
1550
for record, node, prefix_refs, items in \
1551
self._read_nodes_from_store(self._old_root_keys):
1552
# Uninteresting node
1553
prefix_refs = [p_r for p_r in prefix_refs
1554
if p_r[1] not in all_old_chks]
1555
new_refs = [p_r[1] for p_r in prefix_refs]
1556
all_old_chks.update(new_refs)
1557
# TODO: This might be a good time to turn items into StaticTuple
1558
# instances and possibly intern them. However, this does not
1559
# impact 'initial branch' performance, so I'm not worrying
1561
self._all_old_items.update(items)
1562
# Queue up the uninteresting references
1563
# Don't actually put them in the 'to-read' queue until we have
1564
# finished checking the interesting references
1565
old_chks_to_enqueue.extend(prefix_refs)
1566
return old_chks_to_enqueue
1568
def _enqueue_old(self, new_prefixes, old_chks_to_enqueue):
1569
# At this point, we have read all the uninteresting and interesting
1570
# items, so we can queue up the uninteresting stuff, knowing that we've
1571
# handled the interesting ones
1572
for prefix, ref in old_chks_to_enqueue:
1573
not_interesting = True
1574
for i in xrange(len(prefix), 0, -1):
1575
if prefix[:i] in new_prefixes:
1576
not_interesting = False
1579
# This prefix is not part of the remaining 'interesting set'
1581
self._old_queue.append(ref)
1583
def _read_all_roots(self):
1584
"""Read the root pages.
1586
This is structured as a generator, so that the root records can be
1587
yielded up to whoever needs them without any buffering.
1589
# This is the bootstrap phase
1590
if not self._old_root_keys:
1591
# With no old_root_keys we can just shortcut and be ready
1592
# for _flush_new_queue
1593
self._new_queue = list(self._new_root_keys)
1595
old_chks_to_enqueue = self._read_old_roots()
1596
# filter out any root keys that are already known to be uninteresting
1597
new_keys = set(self._new_root_keys).difference(self._all_old_chks)
1598
# These are prefixes that are present in new_keys that we are
1600
new_prefixes = set()
1601
# We are about to yield all of these, so we don't want them getting
1602
# added a second time
1603
processed_new_refs = self._processed_new_refs
1604
processed_new_refs.update(new_keys)
1605
for record, node, prefix_refs, items in \
1606
self._read_nodes_from_store(new_keys):
1607
# At this level, we now know all the uninteresting references
1608
# So we filter and queue up whatever is remaining
1609
prefix_refs = [p_r for p_r in prefix_refs
1610
if p_r[1] not in self._all_old_chks
1611
and p_r[1] not in processed_new_refs]
1612
refs = [p_r[1] for p_r in prefix_refs]
1613
new_prefixes.update([p_r[0] for p_r in prefix_refs])
1614
self._new_queue.extend(refs)
1615
# TODO: We can potentially get multiple items here, however the
1616
# current design allows for this, as callers will do the work
1617
# to make the results unique. We might profile whether we
1618
# gain anything by ensuring unique return values for items
1619
# TODO: This might be a good time to cast to StaticTuple, as
1620
# self._new_item_queue will hold the contents of multiple
1621
# records for an extended lifetime
1622
new_items = [item for item in items
1623
if item not in self._all_old_items]
1624
self._new_item_queue.extend(new_items)
1625
new_prefixes.update([self._search_key_func(item[0])
1626
for item in new_items])
1627
processed_new_refs.update(refs)
1629
# For new_prefixes we have the full length prefixes queued up.
1630
# However, we also need possible prefixes. (If we have a known ref to
1631
# 'ab', then we also need to include 'a'.) So expand the
1632
# new_prefixes to include all shorter prefixes
1633
for prefix in list(new_prefixes):
1634
new_prefixes.update([prefix[:i] for i in xrange(1, len(prefix))])
1635
self._enqueue_old(new_prefixes, old_chks_to_enqueue)
1637
def _flush_new_queue(self):
1638
# No need to maintain the heap invariant anymore, just pull things out
1640
refs = set(self._new_queue)
1641
self._new_queue = []
1642
# First pass, flush all interesting items and convert to using direct refs
1643
all_old_chks = self._all_old_chks
1644
processed_new_refs = self._processed_new_refs
1645
all_old_items = self._all_old_items
1646
new_items = [item for item in self._new_item_queue
1647
if item not in all_old_items]
1648
self._new_item_queue = []
1650
yield None, new_items
1651
refs = refs.difference(all_old_chks)
1652
processed_new_refs.update(refs)
1654
# TODO: Using a SimpleSet for self._processed_new_refs and
1655
# saved as much as 10MB of peak memory. However, it requires
1656
# implementing a non-pyrex version.
1658
next_refs_update = next_refs.update
1659
# Inlining _read_nodes_from_store improves 'bzr branch bzr.dev'
1660
# from 1m54s to 1m51s. Consider it.
1661
for record, _, p_refs, items in self._read_nodes_from_store(refs):
1663
# using the 'if' check saves about 145s => 141s, when
1664
# streaming initial branch of Launchpad data.
1665
items = [item for item in items
1666
if item not in all_old_items]
1668
next_refs_update([p_r[1] for p_r in p_refs])
1670
# set1.difference(set/dict) walks all of set1, and checks if it
1671
# exists in 'other'.
1672
# set1.difference(iterable) walks all of iterable, and does a
1673
# 'difference_update' on a clone of set1. Pick wisely based on the
1674
# expected sizes of objects.
1675
# in our case it is expected that 'new_refs' will always be quite
1677
next_refs = next_refs.difference(all_old_chks)
1678
next_refs = next_refs.difference(processed_new_refs)
1679
processed_new_refs.update(next_refs)
1682
def _process_next_old(self):
1683
# Since we don't filter uninteresting any further than during
1684
# _read_all_roots, process the whole queue in a single pass.
1685
refs = self._old_queue
1686
self._old_queue = []
1687
all_old_chks = self._all_old_chks
1688
for record, _, prefix_refs, items in self._read_nodes_from_store(refs):
1689
# TODO: Use StaticTuple here?
1690
self._all_old_items.update(items)
1691
refs = [r for _,r in prefix_refs if r not in all_old_chks]
1692
self._old_queue.extend(refs)
1693
all_old_chks.update(refs)
1695
def _process_queues(self):
1696
while self._old_queue:
1697
self._process_next_old()
1698
return self._flush_new_queue()
1701
for record in self._read_all_roots():
1703
for record, items in self._process_queues():
1707
def iter_interesting_nodes(store, interesting_root_keys,
1708
uninteresting_root_keys, pb=None):
1709
"""Given root keys, find interesting nodes.
1711
Evaluate nodes referenced by interesting_root_keys. Ones that are also
1712
referenced from uninteresting_root_keys are not considered interesting.
1714
:param interesting_root_keys: keys which should be part of the
1715
"interesting" nodes (which will be yielded)
1716
:param uninteresting_root_keys: keys which should be filtered out of the
1719
(interesting record, {interesting key:values})
1721
iterator = CHKMapDifference(store, interesting_root_keys,
1722
uninteresting_root_keys,
1723
search_key_func=store._search_key_func,
1725
return iterator.process()
1729
from bzrlib._chk_map_pyx import (
1732
_deserialise_leaf_node,
1733
_deserialise_internal_node,
1735
except ImportError, e:
1736
osutils.failed_to_load_extension(e)
1737
from bzrlib._chk_map_py import (
1740
_deserialise_leaf_node,
1741
_deserialise_internal_node,
1743
search_key_registry.register('hash-16-way', _search_key_16)
1744
search_key_registry.register('hash-255-way', _search_key_255)
1747
def _check_key(key):
1748
"""Helper function to assert that a key is properly formatted.
1750
This generally shouldn't be used in production code, but it can be helpful
1753
if type(key) is not StaticTuple:
1754
raise TypeError('key %r is not StaticTuple but %s' % (key, type(key)))
1756
raise ValueError('key %r should have length 1, not %d' % (key, len(key),))
1757
if type(key[0]) is not str:
1758
raise TypeError('key %r should hold a str, not %r'
1759
% (key, type(key[0])))
1760
if not key[0].startswith('sha1:'):
1761
raise ValueError('key %r should point to a sha1:' % (key,))
added
removed
bzrlib/chk_map.py
