~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/chk_map.py

  • Committer: Canonical.com Patch Queue Manager
  • Date: 2009-04-09 20:23:07 UTC
  • mfrom: (4265.1.4 bbc-merge)
  • Revision ID: pqm@pqm.ubuntu.com-20090409202307-n0depb16qepoe21o
(jam) Change _fetch_uses_deltas = False for CHK repos until we can
        write a better fix.

Show diffs side-by-side

added added

removed removed

Lines of Context:
1
 
# Copyright (C) 2008-2011 Canonical Ltd
 
1
# Copyright (C) 2008 Canonical Ltd
2
2
#
3
3
# This program is free software; you can redistribute it and/or modify
4
4
# it under the terms of the GNU General Public License as published by
26
26
 
27
27
Updates to a CHKMap are done preferentially via the apply_delta method, to
28
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.
 
29
possible and supported. All changes via map/unmap are buffered in memory until
 
30
the _save method is called to force serialisation of the tree. apply_delta
 
31
performs a _save implicitly.
32
32
 
33
33
TODO:
34
34
-----
38
38
"""
39
39
 
40
40
import heapq
41
 
import threading
 
41
import time
42
42
 
43
43
from bzrlib import lazy_import
44
44
lazy_import.lazy_import(globals(), """
45
 
from bzrlib import (
46
 
    errors,
47
 
    )
 
45
from bzrlib import versionedfile
48
46
""")
49
47
from bzrlib import (
50
48
    errors,
51
49
    lru_cache,
52
50
    osutils,
53
51
    registry,
54
 
    static_tuple,
55
52
    trace,
56
53
    )
57
 
from bzrlib.static_tuple import StaticTuple
58
54
 
59
55
# approx 4MB
60
56
# If each line is 50 bytes, and you have 255 internal pages, with 255-way fan
61
57
# out, it takes 3.1MB to cache the layer.
62
58
_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
65
 
#   cache thrashing.
66
 
# - we avoid locking on every cache lookup.
67
 
_thread_caches = threading.local()
68
 
# The page cache.
69
 
_thread_caches.page_cache = None
70
 
 
71
 
def _get_cache():
72
 
    """Get the per-thread page cache.
73
 
 
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.
76
 
    """
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
82
 
    return page_cache
83
 
 
84
 
 
85
 
def clear_cache():
86
 
    _get_cache().clear()
87
 
 
 
59
# We are caching bytes so len(value) is perfectly accurate
 
60
_page_cache = lru_cache.LRUSizeCache(_PAGE_CACHE_SIZE)
88
61
 
89
62
# If a ChildNode falls below this many bytes, we check for a remap
90
63
_INTERESTING_NEW_SIZE = 50
91
64
# If a ChildNode shrinks by more than this amount, we check for a remap
92
65
_INTERESTING_SHRINKAGE_LIMIT = 20
 
66
# If we delete more than this many nodes applying a delta, we check for a remap
 
67
_INTERESTING_DELETES_LIMIT = 5
93
68
 
94
69
 
95
70
def _search_key_plain(key):
104
79
class CHKMap(object):
105
80
    """A persistent map from string to string backed by a CHK store."""
106
81
 
107
 
    __slots__ = ('_store', '_root_node', '_search_key_func')
108
 
 
109
82
    def __init__(self, store, root_key, search_key_func=None):
110
83
        """Create a CHKMap object.
111
84
 
133
106
            into the map; if old_key is not None, then the old mapping
134
107
            of old_key is removed.
135
108
        """
136
 
        has_deletes = False
137
 
        # Check preconditions first.
138
 
        as_st = StaticTuple.from_sequence
139
 
        new_items = set([as_st(key) for (old, key, value) in delta
140
 
                         if key is not None and old is None])
141
 
        existing_new = list(self.iteritems(key_filter=new_items))
142
 
        if existing_new:
143
 
            raise errors.InconsistentDeltaDelta(delta,
144
 
                "New items are already in the map %r." % existing_new)
145
 
        # Now apply changes.
 
109
        delete_count = 0
146
110
        for old, new, value in delta:
147
111
            if old is not None and old != new:
148
112
                self.unmap(old, check_remap=False)
149
 
                has_deletes = True
 
113
                delete_count += 1
150
114
        for old, new, value in delta:
151
115
            if new is not None:
152
116
                self.map(new, value)
153
 
        if has_deletes:
 
117
        if delete_count > _INTERESTING_DELETES_LIMIT:
 
118
            trace.mutter("checking remap as %d deletions", delete_count)
154
119
            self._check_remap()
155
120
        return self._save()
156
121
 
157
122
    def _ensure_root(self):
158
123
        """Ensure that the root node is an object not a key."""
159
 
        if type(self._root_node) is StaticTuple:
 
124
        if type(self._root_node) == tuple:
160
125
            # Demand-load the root
161
126
            self._root_node = self._get_node(self._root_node)
162
127
 
170
135
        :param node: A tuple key or node object.
171
136
        :return: A node object.
172
137
        """
173
 
        if type(node) is StaticTuple:
 
138
        if type(node) == tuple:
174
139
            bytes = self._read_bytes(node)
175
140
            return _deserialise(bytes, node,
176
141
                search_key_func=self._search_key_func)
179
144
 
180
145
    def _read_bytes(self, key):
181
146
        try:
182
 
            return _get_cache()[key]
 
147
            return _page_cache[key]
183
148
        except KeyError:
184
149
            stream = self._store.get_record_stream([key], 'unordered', True)
185
150
            bytes = stream.next().get_bytes_as('fulltext')
186
 
            _get_cache()[key] = bytes
 
151
            _page_cache[key] = bytes
187
152
            return bytes
188
153
 
189
154
    def _dump_tree(self, include_keys=False):
217
182
            for key, value in sorted(node._items.iteritems()):
218
183
                # Don't use prefix nor indent here to line up when used in
219
184
                # tests in conjunction with assertEqualDiff
220
 
                result.append('      %r %r' % (tuple(key), value))
 
185
                result.append('      %r %r' % (key, value))
221
186
        return result
222
187
 
223
188
    @classmethod
238
203
            multiple pages.
239
204
        :return: The root chk of the resulting CHKMap.
240
205
        """
241
 
        root_key = klass._create_directly(store, initial_value,
242
 
            maximum_size=maximum_size, key_width=key_width,
243
 
            search_key_func=search_key_func)
244
 
        if type(root_key) is not StaticTuple:
245
 
            raise AssertionError('we got a %s instead of a StaticTuple'
246
 
                                 % (type(root_key),))
247
 
        return root_key
248
 
 
249
 
    @classmethod
250
 
    def _create_via_map(klass, store, initial_value, maximum_size=0,
251
 
                        key_width=1, search_key_func=None):
252
 
        result = klass(store, None, search_key_func=search_key_func)
 
206
        result = CHKMap(store, None, search_key_func=search_key_func)
253
207
        result._root_node.set_maximum_size(maximum_size)
254
208
        result._root_node._key_width = key_width
255
209
        delta = []
256
210
        for key, value in initial_value.items():
257
211
            delta.append((None, key, value))
258
 
        root_key = result.apply_delta(delta)
259
 
        return root_key
260
 
 
261
 
    @classmethod
262
 
    def _create_directly(klass, store, initial_value, maximum_size=0,
263
 
                         key_width=1, search_key_func=None):
264
 
        node = LeafNode(search_key_func=search_key_func)
265
 
        node.set_maximum_size(maximum_size)
266
 
        node._key_width = key_width
267
 
        as_st = StaticTuple.from_sequence
268
 
        node._items = dict([(as_st(key), val) for key, val
269
 
                                               in initial_value.iteritems()])
270
 
        node._raw_size = sum([node._key_value_len(key, value)
271
 
                              for key,value in node._items.iteritems()])
272
 
        node._len = len(node._items)
273
 
        node._compute_search_prefix()
274
 
        node._compute_serialised_prefix()
275
 
        if (node._len > 1
276
 
            and maximum_size
277
 
            and node._current_size() > maximum_size):
278
 
            prefix, node_details = node._split(store)
279
 
            if len(node_details) == 1:
280
 
                raise AssertionError('Failed to split using node._split')
281
 
            node = InternalNode(prefix, search_key_func=search_key_func)
282
 
            node.set_maximum_size(maximum_size)
283
 
            node._key_width = key_width
284
 
            for split, subnode in node_details:
285
 
                node.add_node(split, subnode)
286
 
        keys = list(node.serialise(store))
287
 
        return keys[-1]
 
212
        return result.apply_delta(delta)
288
213
 
289
214
    def iter_changes(self, basis):
290
215
        """Iterate over the changes between basis and self.
510
435
    def iteritems(self, key_filter=None):
511
436
        """Iterate over the entire CHKMap's contents."""
512
437
        self._ensure_root()
513
 
        if key_filter is not None:
514
 
            as_st = StaticTuple.from_sequence
515
 
            key_filter = [as_st(key) for key in key_filter]
516
438
        return self._root_node.iteritems(self._store, key_filter=key_filter)
517
439
 
518
440
    def key(self):
519
441
        """Return the key for this map."""
520
 
        if type(self._root_node) is StaticTuple:
 
442
        if type(self._root_node) is tuple:
521
443
            return self._root_node
522
444
        else:
523
445
            return self._root_node._key
527
449
        return len(self._root_node)
528
450
 
529
451
    def map(self, key, value):
530
 
        """Map a key tuple to value.
531
 
        
532
 
        :param key: A key to map.
533
 
        :param value: The value to assign to key.
534
 
        """
535
 
        key = StaticTuple.from_sequence(key)
 
452
        """Map a key tuple to value."""
536
453
        # Need a root object.
537
454
        self._ensure_root()
538
455
        prefix, node_details = self._root_node.map(self._store, key, value)
548
465
 
549
466
    def _node_key(self, node):
550
467
        """Get the key for a node whether it's a tuple or node."""
551
 
        if type(node) is tuple:
552
 
            node = StaticTuple.from_sequence(node)
553
 
        if type(node) is StaticTuple:
 
468
        if type(node) == tuple:
554
469
            return node
555
470
        else:
556
471
            return node._key
557
472
 
558
473
    def unmap(self, key, check_remap=True):
559
474
        """remove key from the map."""
560
 
        key = StaticTuple.from_sequence(key)
561
475
        self._ensure_root()
562
476
        if type(self._root_node) is InternalNode:
563
477
            unmapped = self._root_node.unmap(self._store, key,
570
484
        """Check if nodes can be collapsed."""
571
485
        self._ensure_root()
572
486
        if type(self._root_node) is InternalNode:
573
 
            self._root_node = self._root_node._check_remap(self._store)
 
487
            self._root_node._check_remap(self._store)
574
488
 
575
489
    def _save(self):
576
490
        """Save the map completely.
577
491
 
578
492
        :return: The key of the root node.
579
493
        """
580
 
        if type(self._root_node) is StaticTuple:
 
494
        if type(self._root_node) == tuple:
581
495
            # Already saved.
582
496
            return self._root_node
583
497
        keys = list(self._root_node.serialise(self._store))
591
505
        adding the header bytes, and without prefix compression.
592
506
    """
593
507
 
594
 
    __slots__ = ('_key', '_len', '_maximum_size', '_key_width',
595
 
                 '_raw_size', '_items', '_search_prefix', '_search_key_func'
596
 
                )
597
 
 
598
508
    def __init__(self, key_width=1):
599
509
        """Create a node.
600
510
 
649
559
        """
650
560
        if key.startswith(prefix):
651
561
            return prefix
652
 
        pos = -1
653
562
        # Is there a better way to do this?
654
563
        for pos, (left, right) in enumerate(zip(prefix, key)):
655
564
            if left != right:
689
598
        the key/value pairs.
690
599
    """
691
600
 
692
 
    __slots__ = ('_common_serialised_prefix',)
693
 
 
694
601
    def __init__(self, search_key_func=None):
695
602
        Node.__init__(self)
696
603
        # All of the keys in this leaf node share this common prefix
697
604
        self._common_serialised_prefix = None
 
605
        self._serialise_key = '\x00'.join
698
606
        if search_key_func is None:
699
607
            self._search_key_func = _search_key_plain
700
608
        else:
738
646
        :param bytes: The bytes of the node.
739
647
        :param key: The key that the serialised node has.
740
648
        """
741
 
        key = static_tuple.expect_static_tuple(key)
742
649
        return _deserialise_leaf_node(bytes, key,
743
650
                                      search_key_func=search_key_func)
744
651
 
856
763
                result[prefix] = node
857
764
            else:
858
765
                node = result[prefix]
859
 
            sub_prefix, node_details = node.map(store, key, value)
860
 
            if len(node_details) > 1:
861
 
                if prefix != sub_prefix:
862
 
                    # This node has been split and is now found via a different
863
 
                    # path
864
 
                    result.pop(prefix)
865
 
                new_node = InternalNode(sub_prefix,
866
 
                    search_key_func=self._search_key_func)
867
 
                new_node.set_maximum_size(self._maximum_size)
868
 
                new_node._key_width = self._key_width
869
 
                for split, node in node_details:
870
 
                    new_node.add_node(split, node)
871
 
                result[prefix] = new_node
 
766
            node.map(store, key, value)
872
767
        return common_prefix, result.items()
873
768
 
874
769
    def map(self, store, key, value):
884
779
                raise AssertionError('%r must be known' % self._search_prefix)
885
780
            return self._search_prefix, [("", self)]
886
781
 
887
 
    _serialise_key = '\x00'.join
888
 
 
889
782
    def serialise(self, store):
890
783
        """Serialise the LeafNode to store.
891
784
 
916
809
            lines.append(serialized[prefix_len:])
917
810
            lines.extend(value_lines)
918
811
        sha1, _, _ = store.add_lines((None,), (), lines)
919
 
        self._key = StaticTuple("sha1:" + sha1,).intern()
 
812
        self._key = ("sha1:" + sha1,)
920
813
        bytes = ''.join(lines)
921
814
        if len(bytes) != self._current_size():
922
815
            raise AssertionError('Invalid _current_size')
923
 
        _get_cache().add(self._key, bytes)
 
816
        _page_cache.add(self._key, bytes)
924
817
        return [self._key]
925
818
 
926
819
    def refs(self):
990
883
        LeafNode or InternalNode.
991
884
    """
992
885
 
993
 
    __slots__ = ('_node_width',)
994
 
 
995
886
    def __init__(self, prefix='', search_key_func=None):
996
887
        Node.__init__(self)
997
888
        # The size of an internalnode with default values and no children.
1039
930
        :param key: The key that the serialised node has.
1040
931
        :return: An InternalNode instance.
1041
932
        """
1042
 
        key = static_tuple.expect_static_tuple(key)
1043
933
        return _deserialise_internal_node(bytes, key,
1044
934
                                          search_key_func=search_key_func)
1045
935
 
1064
954
        # prefix is the key in self._items to use, key_filter is the key_filter
1065
955
        # entries that would match this node
1066
956
        keys = {}
1067
 
        shortcut = False
1068
957
        if key_filter is None:
1069
 
            # yielding all nodes, yield whatever we have, and queue up a read
1070
 
            # for whatever we are missing
1071
 
            shortcut = True
1072
958
            for prefix, node in self._items.iteritems():
1073
 
                if node.__class__ is StaticTuple:
 
959
                if type(node) == tuple:
1074
960
                    keys[node] = (prefix, None)
1075
961
                else:
1076
962
                    yield node, None
1077
 
        elif len(key_filter) == 1:
1078
 
            # Technically, this path could also be handled by the first check
1079
 
            # in 'self._node_width' in length_filters. However, we can handle
1080
 
            # this case without spending any time building up the
1081
 
            # prefix_to_keys, etc state.
1082
 
 
1083
 
            # This is a bit ugly, but TIMEIT showed it to be by far the fastest
1084
 
            # 0.626us   list(key_filter)[0]
1085
 
            #       is a func() for list(), 2 mallocs, and a getitem
1086
 
            # 0.489us   [k for k in key_filter][0]
1087
 
            #       still has the mallocs, avoids the func() call
1088
 
            # 0.350us   iter(key_filter).next()
1089
 
            #       has a func() call, and mallocs an iterator
1090
 
            # 0.125us   for key in key_filter: pass
1091
 
            #       no func() overhead, might malloc an iterator
1092
 
            # 0.105us   for key in key_filter: break
1093
 
            #       no func() overhead, might malloc an iterator, probably
1094
 
            #       avoids checking an 'else' clause as part of the for
1095
 
            for key in key_filter:
1096
 
                break
1097
 
            search_prefix = self._search_prefix_filter(key)
1098
 
            if len(search_prefix) == self._node_width:
1099
 
                # This item will match exactly, so just do a dict lookup, and
1100
 
                # see what we can return
1101
 
                shortcut = True
1102
 
                try:
1103
 
                    node = self._items[search_prefix]
1104
 
                except KeyError:
1105
 
                    # A given key can only match 1 child node, if it isn't
1106
 
                    # there, then we can just return nothing
1107
 
                    return
1108
 
                if node.__class__ is StaticTuple:
1109
 
                    keys[node] = (search_prefix, [key])
1110
 
                else:
1111
 
                    # This is loaded, and the only thing that can match,
1112
 
                    # return
1113
 
                    yield node, [key]
1114
 
                    return
1115
 
        if not shortcut:
1116
 
            # First, convert all keys into a list of search prefixes
1117
 
            # Aggregate common prefixes, and track the keys they come from
 
963
        else:
 
964
            # XXX defaultdict ?
1118
965
            prefix_to_keys = {}
1119
966
            length_filters = {}
1120
967
            for key in key_filter:
1121
 
                search_prefix = self._search_prefix_filter(key)
 
968
                search_key = self._search_prefix_filter(key)
1122
969
                length_filter = length_filters.setdefault(
1123
 
                                    len(search_prefix), set())
1124
 
                length_filter.add(search_prefix)
1125
 
                prefix_to_keys.setdefault(search_prefix, []).append(key)
1126
 
 
1127
 
            if (self._node_width in length_filters
1128
 
                and len(length_filters) == 1):
1129
 
                # all of the search prefixes match exactly _node_width. This
1130
 
                # means that everything is an exact match, and we can do a
1131
 
                # lookup into self._items, rather than iterating over the items
1132
 
                # dict.
1133
 
                search_prefixes = length_filters[self._node_width]
1134
 
                for search_prefix in search_prefixes:
1135
 
                    try:
1136
 
                        node = self._items[search_prefix]
1137
 
                    except KeyError:
1138
 
                        # We can ignore this one
1139
 
                        continue
1140
 
                    node_key_filter = prefix_to_keys[search_prefix]
1141
 
                    if node.__class__ is StaticTuple:
1142
 
                        keys[node] = (search_prefix, node_key_filter)
 
970
                                    len(search_key), set())
 
971
                length_filter.add(search_key)
 
972
                prefix_to_keys.setdefault(search_key, []).append(key)
 
973
            length_filters = length_filters.items()
 
974
            for prefix, node in self._items.iteritems():
 
975
                node_key_filter = []
 
976
                for length, length_filter in length_filters:
 
977
                    sub_prefix = prefix[:length]
 
978
                    if sub_prefix in length_filter:
 
979
                        node_key_filter.extend(prefix_to_keys[sub_prefix])
 
980
                if node_key_filter: # this key matched something, yield it
 
981
                    if type(node) == tuple:
 
982
                        keys[node] = (prefix, node_key_filter)
1143
983
                    else:
1144
984
                        yield node, node_key_filter
1145
 
            else:
1146
 
                # The slow way. We walk every item in self._items, and check to
1147
 
                # see if there are any matches
1148
 
                length_filters = length_filters.items()
1149
 
                for prefix, node in self._items.iteritems():
1150
 
                    node_key_filter = []
1151
 
                    for length, length_filter in length_filters:
1152
 
                        sub_prefix = prefix[:length]
1153
 
                        if sub_prefix in length_filter:
1154
 
                            node_key_filter.extend(prefix_to_keys[sub_prefix])
1155
 
                    if node_key_filter: # this key matched something, yield it
1156
 
                        if node.__class__ is StaticTuple:
1157
 
                            keys[node] = (prefix, node_key_filter)
1158
 
                        else:
1159
 
                            yield node, node_key_filter
1160
985
        if keys:
1161
986
            # Look in the page cache for some more bytes
1162
987
            found_keys = set()
1163
988
            for key in keys:
1164
989
                try:
1165
 
                    bytes = _get_cache()[key]
 
990
                    bytes = _page_cache[key]
1166
991
                except KeyError:
1167
992
                    continue
1168
993
                else:
1193
1018
                    prefix, node_key_filter = keys[record.key]
1194
1019
                    node_and_filters.append((node, node_key_filter))
1195
1020
                    self._items[prefix] = node
1196
 
                    _get_cache().add(record.key, bytes)
 
1021
                    _page_cache.add(record.key, bytes)
1197
1022
                for info in node_and_filters:
1198
1023
                    yield info
1199
1024
 
1291
1116
        :return: An iterable of the keys inserted by this operation.
1292
1117
        """
1293
1118
        for node in self._items.itervalues():
1294
 
            if type(node) is StaticTuple:
 
1119
            if type(node) == tuple:
1295
1120
                # Never deserialised.
1296
1121
                continue
1297
1122
            if node._key is not None:
1308
1133
        lines.append('%s\n' % (self._search_prefix,))
1309
1134
        prefix_len = len(self._search_prefix)
1310
1135
        for prefix, node in sorted(self._items.items()):
1311
 
            if type(node) is StaticTuple:
 
1136
            if type(node) == tuple:
1312
1137
                key = node[0]
1313
1138
            else:
1314
1139
                key = node._key[0]
1318
1143
                    % (serialised, self._search_prefix))
1319
1144
            lines.append(serialised[prefix_len:])
1320
1145
        sha1, _, _ = store.add_lines((None,), (), lines)
1321
 
        self._key = StaticTuple("sha1:" + sha1,).intern()
1322
 
        _get_cache().add(self._key, ''.join(lines))
 
1146
        self._key = ("sha1:" + sha1,)
 
1147
        _page_cache.add(self._key, ''.join(lines))
1323
1148
        yield self._key
1324
1149
 
1325
1150
    def _search_key(self, key):
1353
1178
            raise AssertionError("unserialised nodes have no refs.")
1354
1179
        refs = []
1355
1180
        for value in self._items.itervalues():
1356
 
            if type(value) is StaticTuple:
 
1181
            if type(value) == tuple:
1357
1182
                refs.append(value)
1358
1183
            else:
1359
1184
                refs.append(value.key())
1369
1194
        return self._search_prefix
1370
1195
 
1371
1196
    def unmap(self, store, key, check_remap=True):
1372
 
        """Remove key from this node and its children."""
 
1197
        """Remove key from this node and it's children."""
1373
1198
        if not len(self._items):
1374
1199
            raise AssertionError("can't unmap in an empty InternalNode.")
1375
1200
        children = [node for node, _
1460
1285
    return node
1461
1286
 
1462
1287
 
1463
 
class CHKMapDifference(object):
1464
 
    """Iterate the stored pages and key,value pairs for (new - old).
1465
 
 
1466
 
    This class provides a generator over the stored CHK pages and the
1467
 
    (key, value) pairs that are in any of the new maps and not in any of the
1468
 
    old maps.
1469
 
 
1470
 
    Note that it may yield chk pages that are common (especially root nodes),
1471
 
    but it won't yield (key,value) pairs that are common.
1472
 
    """
1473
 
 
1474
 
    def __init__(self, store, new_root_keys, old_root_keys,
1475
 
                 search_key_func, pb=None):
1476
 
        # TODO: Should we add a StaticTuple barrier here? It would be nice to
1477
 
        #       force callers to use StaticTuple, because there will often be
1478
 
        #       lots of keys passed in here. And even if we cast it locally,
1479
 
        #       that just meanst that we will have *both* a StaticTuple and a
1480
 
        #       tuple() in memory, referring to the same object. (so a net
1481
 
        #       increase in memory, not a decrease.)
1482
 
        self._store = store
1483
 
        self._new_root_keys = new_root_keys
1484
 
        self._old_root_keys = old_root_keys
1485
 
        self._pb = pb
1486
 
        # All uninteresting chks that we have seen. By the time they are added
1487
 
        # here, they should be either fully ignored, or queued up for
1488
 
        # processing
1489
 
        # TODO: This might grow to a large size if there are lots of merge
1490
 
        #       parents, etc. However, it probably doesn't scale to O(history)
1491
 
        #       like _processed_new_refs does.
1492
 
        self._all_old_chks = set(self._old_root_keys)
1493
 
        # All items that we have seen from the old_root_keys
1494
 
        self._all_old_items = set()
1495
 
        # These are interesting items which were either read, or already in the
1496
 
        # interesting queue (so we don't need to walk them again)
1497
 
        # TODO: processed_new_refs becomes O(all_chks), consider switching to
1498
 
        #       SimpleSet here.
1499
 
        self._processed_new_refs = set()
1500
 
        self._search_key_func = search_key_func
1501
 
 
1502
 
        # The uninteresting and interesting nodes to be searched
1503
 
        self._old_queue = []
1504
 
        self._new_queue = []
1505
 
        # Holds the (key, value) items found when processing the root nodes,
1506
 
        # waiting for the uninteresting nodes to be walked
1507
 
        self._new_item_queue = []
1508
 
        self._state = None
1509
 
 
1510
 
    def _read_nodes_from_store(self, keys):
1511
 
        # We chose not to use _get_cache(), because we think in
1512
 
        # terms of records to be yielded. Also, we expect to touch each page
1513
 
        # only 1 time during this code. (We may want to evaluate saving the
1514
 
        # raw bytes into the page cache, which would allow a working tree
1515
 
        # update after the fetch to not have to read the bytes again.)
1516
 
        as_st = StaticTuple.from_sequence
1517
 
        stream = self._store.get_record_stream(keys, 'unordered', True)
1518
 
        for record in stream:
1519
 
            if self._pb is not None:
1520
 
                self._pb.tick()
1521
 
            if record.storage_kind == 'absent':
1522
 
                raise errors.NoSuchRevision(self._store, record.key)
 
1288
def _find_children_info(store, interesting_keys, uninteresting_keys, pb):
 
1289
    """Read the associated records, and determine what is interesting."""
 
1290
    uninteresting_keys = set(uninteresting_keys)
 
1291
    chks_to_read = uninteresting_keys.union(interesting_keys)
 
1292
    next_uninteresting = set()
 
1293
    next_interesting = set()
 
1294
    uninteresting_items = set()
 
1295
    interesting_items = set()
 
1296
    interesting_to_yield = []
 
1297
    for record in store.get_record_stream(chks_to_read, 'unordered', True):
 
1298
        # records_read.add(record.key())
 
1299
        if pb is not None:
 
1300
            pb.tick()
 
1301
        bytes = record.get_bytes_as('fulltext')
 
1302
        # We don't care about search_key_func for this code, because we only
 
1303
        # care about external references.
 
1304
        node = _deserialise(bytes, record.key, search_key_func=None)
 
1305
        if record.key in uninteresting_keys:
 
1306
            if type(node) is InternalNode:
 
1307
                next_uninteresting.update(node.refs())
 
1308
            else:
 
1309
                # We know we are at a LeafNode, so we can pass None for the
 
1310
                # store
 
1311
                uninteresting_items.update(node.iteritems(None))
 
1312
        else:
 
1313
            interesting_to_yield.append(record.key)
 
1314
            if type(node) is InternalNode:
 
1315
                next_interesting.update(node.refs())
 
1316
            else:
 
1317
                interesting_items.update(node.iteritems(None))
 
1318
    return (next_uninteresting, uninteresting_items,
 
1319
            next_interesting, interesting_to_yield, interesting_items)
 
1320
 
 
1321
 
 
1322
def _find_all_uninteresting(store, interesting_root_keys,
 
1323
                            uninteresting_root_keys, pb):
 
1324
    """Determine the full set of uninteresting keys."""
 
1325
    # What about duplicates between interesting_root_keys and
 
1326
    # uninteresting_root_keys?
 
1327
    if not uninteresting_root_keys:
 
1328
        # Shortcut case. We know there is nothing uninteresting to filter out
 
1329
        # So we just let the rest of the algorithm do the work
 
1330
        # We know there is nothing uninteresting, and we didn't have to read
 
1331
        # any interesting records yet.
 
1332
        return (set(), set(), set(interesting_root_keys), [], set())
 
1333
    all_uninteresting_chks = set(uninteresting_root_keys)
 
1334
    all_uninteresting_items = set()
 
1335
 
 
1336
    # First step, find the direct children of both the interesting and
 
1337
    # uninteresting set
 
1338
    (uninteresting_keys, uninteresting_items,
 
1339
     interesting_keys, interesting_to_yield,
 
1340
     interesting_items) = _find_children_info(store, interesting_root_keys,
 
1341
                                              uninteresting_root_keys,
 
1342
                                              pb=pb)
 
1343
    all_uninteresting_chks.update(uninteresting_keys)
 
1344
    all_uninteresting_items.update(uninteresting_items)
 
1345
    del uninteresting_items
 
1346
    # Note: Exact matches between interesting and uninteresting do not need
 
1347
    #       to be search further. Non-exact matches need to be searched in case
 
1348
    #       there is a future exact-match
 
1349
    uninteresting_keys.difference_update(interesting_keys)
 
1350
 
 
1351
    # Second, find the full set of uninteresting bits reachable by the
 
1352
    # uninteresting roots
 
1353
    chks_to_read = uninteresting_keys
 
1354
    while chks_to_read:
 
1355
        next_chks = set()
 
1356
        for record in store.get_record_stream(chks_to_read, 'unordered', False):
 
1357
            # TODO: Handle 'absent'
 
1358
            if pb is not None:
 
1359
                pb.tick()
1523
1360
            bytes = record.get_bytes_as('fulltext')
1524
 
            node = _deserialise(bytes, record.key,
1525
 
                                search_key_func=self._search_key_func)
 
1361
            # We don't care about search_key_func for this code, because we
 
1362
            # only care about external references.
 
1363
            node = _deserialise(bytes, record.key, search_key_func=None)
1526
1364
            if type(node) is InternalNode:
1527
 
                # Note we don't have to do node.refs() because we know that
1528
 
                # there are no children that have been pushed into this node
1529
 
                # Note: Using as_st() here seemed to save 1.2MB, which would
1530
 
                #       indicate that we keep 100k prefix_refs around while
1531
 
                #       processing. They *should* be shorter lived than that...
1532
 
                #       It does cost us ~10s of processing time
1533
 
                #prefix_refs = [as_st(item) for item in node._items.iteritems()]
1534
 
                prefix_refs = node._items.items()
1535
 
                items = []
 
1365
                # uninteresting_prefix_chks.update(node._items.iteritems())
 
1366
                chks = node._items.values()
 
1367
                # TODO: We remove the entries that are already in
 
1368
                #       uninteresting_chks ?
 
1369
                next_chks.update(chks)
 
1370
                all_uninteresting_chks.update(chks)
1536
1371
            else:
1537
 
                prefix_refs = []
1538
 
                # Note: We don't use a StaticTuple here. Profiling showed a
1539
 
                #       minor memory improvement (0.8MB out of 335MB peak 0.2%)
1540
 
                #       But a significant slowdown (15s / 145s, or 10%)
1541
 
                items = node._items.items()
1542
 
            yield record, node, prefix_refs, items
1543
 
 
1544
 
    def _read_old_roots(self):
1545
 
        old_chks_to_enqueue = []
1546
 
        all_old_chks = self._all_old_chks
1547
 
        for record, node, prefix_refs, items in \
1548
 
                self._read_nodes_from_store(self._old_root_keys):
1549
 
            # Uninteresting node
1550
 
            prefix_refs = [p_r for p_r in prefix_refs
1551
 
                                if p_r[1] not in all_old_chks]
1552
 
            new_refs = [p_r[1] for p_r in prefix_refs]
1553
 
            all_old_chks.update(new_refs)
1554
 
            # TODO: This might be a good time to turn items into StaticTuple
1555
 
            #       instances and possibly intern them. However, this does not
1556
 
            #       impact 'initial branch' performance, so I'm not worrying
1557
 
            #       about this yet
1558
 
            self._all_old_items.update(items)
1559
 
            # Queue up the uninteresting references
1560
 
            # Don't actually put them in the 'to-read' queue until we have
1561
 
            # finished checking the interesting references
1562
 
            old_chks_to_enqueue.extend(prefix_refs)
1563
 
        return old_chks_to_enqueue
1564
 
 
1565
 
    def _enqueue_old(self, new_prefixes, old_chks_to_enqueue):
1566
 
        # At this point, we have read all the uninteresting and interesting
1567
 
        # items, so we can queue up the uninteresting stuff, knowing that we've
1568
 
        # handled the interesting ones
1569
 
        for prefix, ref in old_chks_to_enqueue:
1570
 
            not_interesting = True
1571
 
            for i in xrange(len(prefix), 0, -1):
1572
 
                if prefix[:i] in new_prefixes:
1573
 
                    not_interesting = False
1574
 
                    break
1575
 
            if not_interesting:
1576
 
                # This prefix is not part of the remaining 'interesting set'
1577
 
                continue
1578
 
            self._old_queue.append(ref)
1579
 
 
1580
 
    def _read_all_roots(self):
1581
 
        """Read the root pages.
1582
 
 
1583
 
        This is structured as a generator, so that the root records can be
1584
 
        yielded up to whoever needs them without any buffering.
1585
 
        """
1586
 
        # This is the bootstrap phase
1587
 
        if not self._old_root_keys:
1588
 
            # With no old_root_keys we can just shortcut and be ready
1589
 
            # for _flush_new_queue
1590
 
            self._new_queue = list(self._new_root_keys)
1591
 
            return
1592
 
        old_chks_to_enqueue = self._read_old_roots()
1593
 
        # filter out any root keys that are already known to be uninteresting
1594
 
        new_keys = set(self._new_root_keys).difference(self._all_old_chks)
1595
 
        # These are prefixes that are present in new_keys that we are
1596
 
        # thinking to yield
1597
 
        new_prefixes = set()
1598
 
        # We are about to yield all of these, so we don't want them getting
1599
 
        # added a second time
1600
 
        processed_new_refs = self._processed_new_refs
1601
 
        processed_new_refs.update(new_keys)
1602
 
        for record, node, prefix_refs, items in \
1603
 
                self._read_nodes_from_store(new_keys):
1604
 
            # At this level, we now know all the uninteresting references
1605
 
            # So we filter and queue up whatever is remaining
1606
 
            prefix_refs = [p_r for p_r in prefix_refs
1607
 
                           if p_r[1] not in self._all_old_chks
1608
 
                              and p_r[1] not in processed_new_refs]
1609
 
            refs = [p_r[1] for p_r in prefix_refs]
1610
 
            new_prefixes.update([p_r[0] for p_r in prefix_refs])
1611
 
            self._new_queue.extend(refs)
1612
 
            # TODO: We can potentially get multiple items here, however the
1613
 
            #       current design allows for this, as callers will do the work
1614
 
            #       to make the results unique. We might profile whether we
1615
 
            #       gain anything by ensuring unique return values for items
1616
 
            # TODO: This might be a good time to cast to StaticTuple, as
1617
 
            #       self._new_item_queue will hold the contents of multiple
1618
 
            #       records for an extended lifetime
1619
 
            new_items = [item for item in items
1620
 
                               if item not in self._all_old_items]
1621
 
            self._new_item_queue.extend(new_items)
1622
 
            new_prefixes.update([self._search_key_func(item[0])
1623
 
                                 for item in new_items])
1624
 
            processed_new_refs.update(refs)
1625
 
            yield record
1626
 
        # For new_prefixes we have the full length prefixes queued up.
1627
 
        # However, we also need possible prefixes. (If we have a known ref to
1628
 
        # 'ab', then we also need to include 'a'.) So expand the
1629
 
        # new_prefixes to include all shorter prefixes
1630
 
        for prefix in list(new_prefixes):
1631
 
            new_prefixes.update([prefix[:i] for i in xrange(1, len(prefix))])
1632
 
        self._enqueue_old(new_prefixes, old_chks_to_enqueue)
1633
 
 
1634
 
    def _flush_new_queue(self):
1635
 
        # No need to maintain the heap invariant anymore, just pull things out
1636
 
        # and process them
1637
 
        refs = set(self._new_queue)
1638
 
        self._new_queue = []
1639
 
        # First pass, flush all interesting items and convert to using direct refs
1640
 
        all_old_chks = self._all_old_chks
1641
 
        processed_new_refs = self._processed_new_refs
1642
 
        all_old_items = self._all_old_items
1643
 
        new_items = [item for item in self._new_item_queue
1644
 
                           if item not in all_old_items]
1645
 
        self._new_item_queue = []
1646
 
        if new_items:
1647
 
            yield None, new_items
1648
 
        refs = refs.difference(all_old_chks)
1649
 
        processed_new_refs.update(refs)
1650
 
        while refs:
1651
 
            # TODO: Using a SimpleSet for self._processed_new_refs and
1652
 
            #       saved as much as 10MB of peak memory. However, it requires
1653
 
            #       implementing a non-pyrex version.
1654
 
            next_refs = set()
1655
 
            next_refs_update = next_refs.update
1656
 
            # Inlining _read_nodes_from_store improves 'bzr branch bzr.dev'
1657
 
            # from 1m54s to 1m51s. Consider it.
1658
 
            for record, _, p_refs, items in self._read_nodes_from_store(refs):
1659
 
                if all_old_items:
1660
 
                    # using the 'if' check saves about 145s => 141s, when
1661
 
                    # streaming initial branch of Launchpad data.
1662
 
                    items = [item for item in items
1663
 
                             if item not in all_old_items]
1664
 
                yield record, items
1665
 
                next_refs_update([p_r[1] for p_r in p_refs])
1666
 
                del p_refs
1667
 
            # set1.difference(set/dict) walks all of set1, and checks if it
1668
 
            # exists in 'other'.
1669
 
            # set1.difference(iterable) walks all of iterable, and does a
1670
 
            # 'difference_update' on a clone of set1. Pick wisely based on the
1671
 
            # expected sizes of objects.
1672
 
            # in our case it is expected that 'new_refs' will always be quite
1673
 
            # small.
1674
 
            next_refs = next_refs.difference(all_old_chks)
1675
 
            next_refs = next_refs.difference(processed_new_refs)
1676
 
            processed_new_refs.update(next_refs)
1677
 
            refs = next_refs
1678
 
 
1679
 
    def _process_next_old(self):
1680
 
        # Since we don't filter uninteresting any further than during
1681
 
        # _read_all_roots, process the whole queue in a single pass.
1682
 
        refs = self._old_queue
1683
 
        self._old_queue = []
1684
 
        all_old_chks = self._all_old_chks
1685
 
        for record, _, prefix_refs, items in self._read_nodes_from_store(refs):
1686
 
            # TODO: Use StaticTuple here?
1687
 
            self._all_old_items.update(items)
1688
 
            refs = [r for _,r in prefix_refs if r not in all_old_chks]
1689
 
            self._old_queue.extend(refs)
1690
 
            all_old_chks.update(refs)
1691
 
 
1692
 
    def _process_queues(self):
1693
 
        while self._old_queue:
1694
 
            self._process_next_old()
1695
 
        return self._flush_new_queue()
1696
 
 
1697
 
    def process(self):
1698
 
        for record in self._read_all_roots():
1699
 
            yield record, []
1700
 
        for record, items in self._process_queues():
1701
 
            yield record, items
 
1372
                all_uninteresting_items.update(node._items.iteritems())
 
1373
        chks_to_read = next_chks
 
1374
    return (all_uninteresting_chks, all_uninteresting_items,
 
1375
            interesting_keys, interesting_to_yield, interesting_items)
1702
1376
 
1703
1377
 
1704
1378
def iter_interesting_nodes(store, interesting_root_keys,
1715
1389
    :return: Yield
1716
1390
        (interesting record, {interesting key:values})
1717
1391
    """
1718
 
    iterator = CHKMapDifference(store, interesting_root_keys,
1719
 
                                uninteresting_root_keys,
1720
 
                                search_key_func=store._search_key_func,
1721
 
                                pb=pb)
1722
 
    return iterator.process()
 
1392
    # TODO: consider that it may be more memory efficient to use the 20-byte
 
1393
    #       sha1 string, rather than tuples of hexidecimal sha1 strings.
 
1394
    # TODO: Try to factor out a lot of the get_record_stream() calls into a
 
1395
    #       helper function similar to _read_bytes. This function should be
 
1396
    #       able to use nodes from the _page_cache as well as actually
 
1397
    #       requesting bytes from the store.
 
1398
 
 
1399
    (all_uninteresting_chks, all_uninteresting_items, interesting_keys,
 
1400
     interesting_to_yield, interesting_items) = _find_all_uninteresting(store,
 
1401
        interesting_root_keys, uninteresting_root_keys, pb)
 
1402
 
 
1403
    # Now that we know everything uninteresting, we can yield information from
 
1404
    # our first request
 
1405
    interesting_items.difference_update(all_uninteresting_items)
 
1406
    interesting_to_yield = set(interesting_to_yield) - all_uninteresting_chks
 
1407
    if interesting_items:
 
1408
        yield None, interesting_items
 
1409
    if interesting_to_yield:
 
1410
        # We request these records again, rather than buffering the root
 
1411
        # records, most likely they are still in the _group_cache anyway.
 
1412
        for record in store.get_record_stream(interesting_to_yield,
 
1413
                                              'unordered', False):
 
1414
            yield record, []
 
1415
    all_uninteresting_chks.update(interesting_to_yield)
 
1416
    interesting_keys.difference_update(all_uninteresting_chks)
 
1417
 
 
1418
    chks_to_read = interesting_keys
 
1419
    counter = 0
 
1420
    while chks_to_read:
 
1421
        next_chks = set()
 
1422
        for record in store.get_record_stream(chks_to_read, 'unordered', False):
 
1423
            counter += 1
 
1424
            if pb is not None:
 
1425
                pb.update('find chk pages', counter)
 
1426
            # TODO: Handle 'absent'?
 
1427
            bytes = record.get_bytes_as('fulltext')
 
1428
            # We don't care about search_key_func for this code, because we
 
1429
            # only care about external references.
 
1430
            node = _deserialise(bytes, record.key, search_key_func=None)
 
1431
            if type(node) is InternalNode:
 
1432
                # all_uninteresting_chks grows large, as it lists all nodes we
 
1433
                # don't want to process (including already seen interesting
 
1434
                # nodes).
 
1435
                # small.difference_update(large) scales O(large), but
 
1436
                # small.difference(large) scales O(small).
 
1437
                # Also, we know we just _deserialised this node, so we can
 
1438
                # access the dict directly.
 
1439
                chks = set(node._items.itervalues()).difference(
 
1440
                            all_uninteresting_chks)
 
1441
                # Is set() and .difference_update better than:
 
1442
                # chks = [chk for chk in node.refs()
 
1443
                #              if chk not in all_uninteresting_chks]
 
1444
                next_chks.update(chks)
 
1445
                # These are now uninteresting everywhere else
 
1446
                all_uninteresting_chks.update(chks)
 
1447
                interesting_items = []
 
1448
            else:
 
1449
                interesting_items = [item for item in node._items.iteritems()
 
1450
                                     if item not in all_uninteresting_items]
 
1451
                # TODO: Do we need to filter out items that we have already
 
1452
                #       seen on other pages? We don't really want to buffer the
 
1453
                #       whole thing, but it does mean that callers need to
 
1454
                #       understand they may get duplicate values.
 
1455
                # all_uninteresting_items.update(interesting_items)
 
1456
            yield record, interesting_items
 
1457
        chks_to_read = next_chks
1723
1458
 
1724
1459
 
1725
1460
try:
1726
1461
    from bzrlib._chk_map_pyx import (
1727
 
        _bytes_to_text_key,
1728
1462
        _search_key_16,
1729
1463
        _search_key_255,
1730
1464
        _deserialise_leaf_node,
1731
1465
        _deserialise_internal_node,
1732
1466
        )
1733
 
except ImportError, e:
1734
 
    osutils.failed_to_load_extension(e)
 
1467
except ImportError:
1735
1468
    from bzrlib._chk_map_py import (
1736
 
        _bytes_to_text_key,
1737
1469
        _search_key_16,
1738
1470
        _search_key_255,
1739
1471
        _deserialise_leaf_node,
1741
1473
        )
1742
1474
search_key_registry.register('hash-16-way', _search_key_16)
1743
1475
search_key_registry.register('hash-255-way', _search_key_255)
1744
 
 
1745
 
 
1746
 
def _check_key(key):
1747
 
    """Helper function to assert that a key is properly formatted.
1748
 
 
1749
 
    This generally shouldn't be used in production code, but it can be helpful
1750
 
    to debug problems.
1751
 
    """
1752
 
    if type(key) is not StaticTuple:
1753
 
        raise TypeError('key %r is not StaticTuple but %s' % (key, type(key)))
1754
 
    if len(key) != 1:
1755
 
        raise ValueError('key %r should have length 1, not %d' % (key, len(key),))
1756
 
    if type(key[0]) is not str:
1757
 
        raise TypeError('key %r should hold a str, not %r'
1758
 
                        % (key, type(key[0])))
1759
 
    if not key[0].startswith('sha1:'):
1760
 
        raise ValueError('key %r should point to a sha1:' % (key,))
1761
 
 
1762